SageMaker Serve

SageMaker Serve#

Model serving and inference capabilities for deploying and managing ML models.

Model Deployment#

Local SageMaker Serve development package.

This __init__.py file imports key modules used by inference scripts to prevent Python module resolution conflicts with external serve.py files.

The imports below “prime” the module cache so that sagemaker.serve is recognized as a package, preventing conflicts when inference scripts import from submodules.

Bases: AIBenchmarkJob

AIBenchmarkJob with a one-shot result reader.

All standard lifecycle methods (refresh, wait, stop, delete) are inherited from the underlying resource; show_result is the only addition.

ai_benchmark_job_arn: str | PipelineVariable | None#

ai_benchmark_job_name: str | PipelineVariable#

ai_benchmark_job_status: str | PipelineVariable | None#

ai_workload_config_identifier: str | PipelineVariable | None#

benchmark_target: AIBenchmarkTarget | None#

creation_time: datetime | None#

end_time: datetime | None#

failure_reason: str | PipelineVariable | None#

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'protected_namespaces': (), 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

network_config: AIBenchmarkNetworkConfig | None#

output_config: AIBenchmarkOutputResult | None#

role_arn: str | PipelineVariable | None#

show_result()[source]#

Download the benchmark output from S3 and return a parsed result.

Returns:: parsed metrics and run metadata. The job must be in a terminal state; show_result calls refresh() once but does not poll.
Return type:: BenchmarkResult

start_time: datetime | None#

tags: List[Tag] | None#

class sagemaker.serve.BenchmarkResult(metrics: ~sagemaker.serve.ai_inference_recommender.result.BenchmarkMetrics, s3_output_location: str, endpoint: str | None = None, workload_config: str | None = None, tool_version: str | None = None, profile: ~typing.Dict[str, ~typing.Any] = <factory>)[source]#

Bases: object

Parsed result of a completed benchmark job.

endpoint: str | None = None#

classmethod from_job(job, *, session: boto3.session.Session | None = None) → BenchmarkResult[source]#

Download and parse the benchmark output for a completed AIBenchmarkJob.

Populates endpoint, workload_config, and tool_version from the job’s BenchmarkTarget and WorkloadConfigIdentifier plus the AIPerf profile metadata so the parsed result is self-describing.

Parameters:

job – An AIBenchmarkJob (or BenchmarkJob re-export) that has reached a terminal state.
session – Optional boto3 session. Defaults to the ambient session.

Returns:

A parsed BenchmarkResult.

Raises:

RuntimeError – if the job has no S3 output location set.

classmethod from_s3(s3_output_location: str, *, session: boto3.session.Session | None = None, endpoint: str | None = None, workload_config: str | None = None) → BenchmarkResult[source]#

Download and parse the benchmark output artifact from S3.

Parameters:

s3_output_location – s3://bucket/prefix/ location written by the benchmark job.
session – Optional boto3 session. Defaults to the ambient session.
endpoint – Optional endpoint identifier to attach to the result. Threaded through by from_job().
workload_config – Optional workload-config identifier to attach. Threaded through by from_job().

Returns:

A parsed BenchmarkResult.

metrics: BenchmarkMetrics#

profile: Dict[str, Any]#

s3_output_location: str#

tool_version: str | None = None#

workload_config: str | None = None#

exception sagemaker.serve.FeatureGatedError(message: str = '', runbook_url: str = 'https://docs.aws.amazon.com/sagemaker/latest/dg/generative-ai-inference-recommendations.html')[source]#

Bases: SageMakerCoreError

Raised when the AI inference recommender feature is not enabled for the account.

fmt = 'The AI inference recommender feature is not enabled for this account. {message} See {runbook_url} for enrollment information.'#

class sagemaker.serve.InferenceFramework(value)[source]#

Bases: str, Enum

Inference framework to benchmark a recommendation against.

LMI = 'LMI'#

VLLM = 'VLLM'#

class sagemaker.serve.InferenceSpec[source]#

Bases: ABC

Abstract base class for holding custom load, invoke and prepare functions.

Provides a skeleton for customization to override the methods load, invoke and prepare.

get_model()[source]#: Return HuggingFace model name for inference spec

abstract invoke(input_object: object, model: object)[source]#

Given model object and input, make inference and return the result.

Parameters:

input_object (object) – The input to model
model (object) – The model object

abstract load(model_dir: str)[source]#

Loads the model stored in model_dir and return the model object.

Parameters:: model_dir (str) – Path to the directory where the model is stored.

postprocess(predictions: object)[source]#: Custom post-processing function

prepare(*args, **kwargs)[source]#: Custom prepare function

preprocess(input_data: object)[source]#: Custom pre-processing function

class sagemaker.serve.ModelBuilder(model: object | str | ~sagemaker.train.model_trainer.ModelTrainer | ~sagemaker.train.base_trainer.BaseTrainer | ~sagemaker.core.resources.TrainingJob | ~sagemaker.core.resources.ModelPackage | ~typing.List[~sagemaker.core.resources.Model] | None = None, model_path: str | None = <factory>, inference_spec: ~sagemaker.serve.spec.inference_spec.InferenceSpec | None = None, schema_builder: ~sagemaker.serve.builder.schema_builder.SchemaBuilder | None = None, modelbuilder_list: ~typing.List[~sagemaker.serve.model_builder.ModelBuilder] | None = None, role_arn: str | None = None, sagemaker_session: ~sagemaker.core.helper.session_helper.Session | None = None, image_uri: str | ~sagemaker.core.helper.pipeline_variable.PipelineVariable | None = None, s3_model_data_url: str | ~sagemaker.core.helper.pipeline_variable.PipelineVariable | ~typing.Dict[str, ~typing.Any] | None = None, source_code: ~sagemaker.core.training.configs.SourceCode | None = None, env_vars: ~typing.Dict[str, str | ~sagemaker.core.helper.pipeline_variable.PipelineVariable] | None = <factory>, model_server: ~sagemaker.serve.utils.types.ModelServer | None = None, model_metadata: ~typing.Dict[str, ~typing.Any] | None = None, log_level: int | None = 10, content_type: str | None = None, accept_type: str | None = None, compute: ~sagemaker.core.training.configs.Compute | None = None, network: ~sagemaker.core.training.configs.Networking | None = None, instance_type: str | None = None, mode: ~sagemaker.serve.mode.function_pointers.Mode | None = Mode.SAGEMAKER_ENDPOINT, shared_libs: ~typing.List[str] = <factory>, dependencies: ~typing.Dict[str, ~typing.Any] | None = <factory>, image_config: ~typing.Dict[str, str | ~sagemaker.core.helper.pipeline_variable.PipelineVariable] | None = None)[source]#

Bases: _InferenceRecommenderMixin, _ModelBuilderServers, _ModelBuilderUtils

Unified interface for building and deploying machine learning models.

ModelBuilder provides a streamlined workflow for preparing and deploying ML models to Amazon SageMaker. It supports multiple frameworks (PyTorch, TensorFlow, HuggingFace, etc.), model servers (TorchServe, TGI, Triton, etc.), and deployment modes (SageMaker endpoints, local containers, in-process).

The typical workflow involves three steps: 1. Initialize ModelBuilder with your model and configuration 2. Call build() to create a deployable Model resource 3. Call deploy() to create an Endpoint resource for inference

Example

>>> from sagemaker.serve.model_builder import ModelBuilder
>>> from sagemaker.serve.mode.function_pointers import Mode
>>>
>>> # Initialize with a trained model
>>> model_builder = ModelBuilder(
...     model=my_pytorch_model,
...     role_arn="arn:aws:iam::123456789012:role/SageMakerRole",
...     instance_type="ml.m5.xlarge"
... )
>>>
>>> # Build the model (creates SageMaker Model resource)
>>> model = model_builder.build()
>>>
>>> # Deploy to endpoint (creates SageMaker Endpoint resource)
>>> endpoint = model_builder.deploy(endpoint_name="my-endpoint")
>>>
>>> # Make predictions
>>> result = endpoint.invoke(data=input_data)

Parameters:

model – The model to deploy. Can be a trained model object, ModelTrainer, TrainingJob, ModelPackage, or JumpStart model ID string. Either model or inference_spec is required.
model_path – Local directory path where model artifacts are stored or will be downloaded.
inference_spec – Custom inference specification with load() and invoke() functions.
schema_builder – Defines input/output schema for serialization and deserialization.
modelbuilder_list – List of ModelBuilder objects for multi-model deployments.
pipeline_models – List of Model objects for creating inference pipelines.
role_arn – IAM role ARN for SageMaker to assume.
sagemaker_session – Session object for managing SageMaker API interactions.
image_uri – Container image URI. Auto-selected if not specified.
s3_model_data_url – S3 URI where model artifacts are stored or will be uploaded.
source_code – Source code configuration for custom inference code.
env_vars – Environment variables to set in the container.
model_server – Model server to use (TORCHSERVE, TGI, TRITON, etc.).
model_metadata – Dictionary to override model metadata (HF_TASK, MLFLOW_MODEL_PATH, etc.).
log_level – Logging level for ModelBuilder operations (default: logging.DEBUG).
content_type – MIME type of input data. Auto-derived from schema_builder if provided.
accept_type – MIME type of output data. Auto-derived from schema_builder if provided.
compute – Compute configuration specifying instance type and count.
network – Network configuration including VPC settings and network isolation.
instance_type – EC2 instance type for deployment (e.g., ‘ml.m5.large’).
mode – Deployment mode (SAGEMAKER_ENDPOINT, LOCAL_CONTAINER, or IN_PROCESS).

Note

ModelBuilder returns sagemaker.core.resources.Model and sagemaker.core.resources.Endpoint objects, not the deprecated PySDK Model and Predictor classes. Use endpoint.invoke() instead of predictor.predict() for inference.

accept_type: str | None = None#

Build a deployable Model instance with ModelBuilder.

Creates a SageMaker Model resource with the appropriate container image, model artifacts, and configuration. This method prepares the model for deployment but does not deploy it to an endpoint. Use the deploy() method to create an endpoint.

Note: This returns a sagemaker.core.resources.Model object, not the deprecated PySDK Model class.

Parameters:

model_name (str, optional) – The name for the SageMaker model. If not specified, a unique name will be generated. (Default: None).
mode (Mode, optional) – The mode of operation. Options are SAGEMAKER_ENDPOINT, LOCAL_CONTAINER, or IN_PROCESS. (Default: None, uses mode from initialization).
role_arn (str, optional) – The IAM role ARN for SageMaker to assume when creating the model and endpoint. (Default: None).
sagemaker_session (Session, optional) – Session object which manages interactions with Amazon SageMaker APIs and any other AWS services needed. If not specified, uses the session from initialization or creates one using the default AWS configuration chain. (Default: None).
region (str, optional) – The AWS region for deployment. If specified and different from the current region, a new session will be created. (Default: None).

Returns:

A sagemaker.core.resources.Model resource: that represents the created SageMaker model, or a ModelBuilder instance for multi-model scenarios.

Return type:

Union[Model, ModelBuilder, None]

Example

>>> model_builder = ModelBuilder(model=my_model, role_arn=role)
>>> model = model_builder.build()  # Creates Model resource
>>> endpoint = model_builder.deploy()  # Creates Endpoint resource
>>> result = endpoint.invoke(data=input_data)

compute: Compute | None = None#

configure_for_torchserve(shared_libs: List[str] | None = None, dependencies: Dict[str, Any] | None = None, image_config: Dict[str, str | PipelineVariable] | None = None) → ModelBuilder[source]#: Configure ModelBuilder for TorchServe deployment.

content_type: str | None = None#

dependencies: Dict[str, Any] | None#

deploy(endpoint_name: str = None, initial_instance_count: int | None = 1, instance_type: str | None = None, wait: bool = True, update_endpoint: bool | None = False, container_timeout_in_seconds: int = 300, inference_config: ServerlessInferenceConfig | AsyncInferenceConfig | BatchTransformInferenceConfig | ResourceRequirements | None = None, custom_orchestrator_instance_type: str = None, custom_orchestrator_initial_instance_count: int = None, use_recommendation: bool | None = None, recommendation_index: int = 0, recommendation_spec_name: str | None = None, auto_approve: bool = False, role: str | None = None, model_name: str | None = None, endpoint_config_name: str | None = None, **kwargs) → Endpoint | LocalEndpoint | Transformer[source]#

Deploy the built model to an Endpoint.

Creates a SageMaker EndpointConfig and deploys an Endpoint resource from the model created by build(). The model must be built before calling deploy().

Note: This returns a sagemaker.core.resources.Endpoint object, not the deprecated PySDK Predictor class. Use endpoint.invoke() to make predictions.

Parameters:

endpoint_name (str) – The name of the endpoint to create. If not specified, a unique endpoint name will be created. (Default: “endpoint”).
initial_instance_count (int, optional) – The initial number of instances to run in the endpoint. Required for instance-based endpoints. (Default: 1).
instance_type (str, optional) – The EC2 instance type to deploy this model to. For example, ‘ml.p2.xlarge’. Required for instance-based endpoints unless using serverless inference. (Default: None).
wait (bool) – Whether the call should wait until the deployment completes. (Default: True).
update_endpoint (bool) – Flag to update the model in an existing Amazon SageMaker endpoint. If True, deploys a new EndpointConfig to an existing endpoint and deletes resources from the previous EndpointConfig. (Default: False).
container_timeout_in_seconds (int) – The timeout value, in seconds, for the container to respond to requests. (Default: 300).
(Union[ServerlessInferenceConfig (inference_config) – BatchTransformInferenceConfig, ResourceRequirements], optional): Unified inference configuration parameter. Can be used instead of individual config parameters. (Default: None).
AsyncInferenceConfig – BatchTransformInferenceConfig, ResourceRequirements], optional): Unified inference configuration parameter. Can be used instead of individual config parameters. (Default: None).

:paramBatchTransformInferenceConfig, ResourceRequirements], optional): Unified inference: configuration parameter. Can be used instead of individual config parameters. (Default: None).

Parameters:

custom_orchestrator_instance_type (str, optional) – Instance type for custom orchestrator deployment. (Default: None).
custom_orchestrator_initial_instance_count (int, optional) – Initial instance count for custom orchestrator deployment. (Default: None).
use_recommendation (bool, optional) – Controls the recommendation deploy path. None (default) deploys the recommendation when a recommendation job is attached, else the built model. False forces the built-model path even if a job is attached. True requires an attached job and errors otherwise.
recommendation_index (int) – Recommendation deploy only. Index of the recommendation row to deploy. (Default: 0, the top-ranked row). Ignored when deploying a normally-built model.
recommendation_spec_name (str, optional) – Recommendation deploy only. Deploy the row with this inference specification name instead of by index. Ignored when deploying a normally-built model.
auto_approve (bool) – Recommendation deploy only. If True, approve the recommendation’s ModelPackage in place when it is not already Approved (bypassing manual-approval governance on its model package group); if False (default), an unapproved package raises. Ignored when deploying a normally-built model.
role (str, optional) – Recommendation deploy only. Execution role ARN for the Model created from the recommendation’s ModelPackage; defaults to the builder’s role_arn. Ignored when deploying a normally-built model (that path uses the builder’s role).
model_name (str, optional) – Recommendation deploy only. Name for the created Model; auto-generated if omitted. Ignored when deploying a normally-built model.
endpoint_config_name (str, optional) – Recommendation deploy only. Name for the created EndpointConfig; auto-generated if omitted. Ignored when deploying a normally-built model.

Returns:

A sagemaker.core.resources.Endpoint: resource representing the deployed endpoint, a LocalEndpoint for local mode, or a Transformer for batch transform inference.

Return type:

Union[Endpoint, LocalEndpoint, Transformer]

Example

>>> model_builder = ModelBuilder(model=my_model, role_arn=role, instance_type="ml.m5.xlarge")
>>> model = model_builder.build()  # Creates Model resource
>>> endpoint = model_builder.deploy(endpoint_name="my-endpoint")  # Creates Endpoint resource
>>> result = endpoint.invoke(data=input_data)  # Make predictions

deploy_local(endpoint_name: str = 'endpoint', container_timeout_in_seconds: int = 300, **kwargs) → LocalEndpoint[source]#

Deploy the built model to local mode for testing.

Deploys the model locally using either LOCAL_CONTAINER mode (runs in a Docker container) or IN_PROCESS mode (runs in the current Python process). This is useful for testing and development before deploying to SageMaker endpoints. The model must be built with mode=Mode.LOCAL_CONTAINER or mode=Mode.IN_PROCESS before calling this method.

Note: This returns a LocalEndpoint object for local inference, not a SageMaker Endpoint resource. Use local_endpoint.invoke() to make predictions.

Parameters:

endpoint_name (str) – The name for the local endpoint. (Default: “endpoint”).
container_timeout_in_seconds (int) – The timeout value, in seconds, for the container to respond to requests. (Default: 300).

Returns:

A LocalEndpoint object for making local predictions.

Return type:

LocalEndpoint

Raises:

ValueError – If the model was not built with LOCAL_CONTAINER or IN_PROCESS mode.

Example

>>> model_builder = ModelBuilder(
...     model=my_model,
...     role_arn=role,
...     mode=Mode.LOCAL_CONTAINER
... )
>>> model = model_builder.build()
>>> local_endpoint = model_builder.deploy_local()
>>> result = local_endpoint.invoke(data=input_data)

display_benchmark_metrics(**kwargs) → None[source]#: Display benchmark metrics for JumpStart models.

enable_network_isolation()[source]#

Whether to enable network isolation when creating this Model

Returns:: If network isolation should be enabled or not.
Return type:: bool

env_vars: Dict[str, str | PipelineVariable] | None#

fetch_endpoint_names_for_base_model() → Set[str][source]#

Fetches endpoint names for the base model.

Returns:: Set of endpoint names for the base model.

classmethod from_jumpstart_config(jumpstart_config: JumpStartConfig, role_arn: str | None = None, compute: Compute | None = None, network: Networking | None = None, image_uri: str | None = None, env_vars: Dict[str, str] | None = None, model_kms_key: str | None = None, resource_requirements: ResourceRequirements | None = None, tolerate_vulnerable_model: bool | None = None, tolerate_deprecated_model: bool | None = None, sagemaker_session: Session | None = None, schema_builder: SchemaBuilder | None = None) → ModelBuilder[source]#

Create a ModelBuilder instance from a JumpStart configuration.

This class method provides a convenient way to create a ModelBuilder for deploying pre-trained models from Amazon SageMaker JumpStart. It automatically retrieves the appropriate model artifacts, container images, and default configurations for the specified JumpStart model.

Parameters:

jumpstart_config (JumpStartConfig) – Configuration object specifying the JumpStart model to use. Must include model_id and optionally model_version and inference_config_name.
role_arn (str, optional) – The IAM role ARN for SageMaker to assume when creating the model and endpoint. If not specified, attempts to use the default SageMaker execution role. (Default: None).
compute (Compute, optional) – Compute configuration specifying instance type and instance count for deployment. For example, Compute(instance_type=’ml.g5.xlarge’, instance_count=1). (Default: None).
network (Networking, optional) – Network configuration including VPC settings and network isolation. For example, Networking(vpc_config={‘Subnets’: […], ‘SecurityGroupIds’: […]}, enable_network_isolation=False). (Default: None).
image_uri (str, optional) – Custom container image URI. If not specified, uses the default JumpStart container image for the model. (Default: None).
env_vars (Dict[str, str], optional) – Environment variables to set in the container. These will be merged with default JumpStart environment variables. (Default: None).
model_kms_key (str, optional) – KMS key ARN used to encrypt model artifacts when uploading to S3. (Default: None).
resource_requirements (ResourceRequirements, optional) – The compute resource requirements for deploying the model to an inference component based endpoint. (Default: None).
tolerate_vulnerable_model (bool, optional) – If True, allows deployment of models with known security vulnerabilities. Use with caution. (Default: None).
tolerate_deprecated_model (bool, optional) – If True, allows deployment of deprecated JumpStart models. (Default: None).
sagemaker_session (Session, optional) – Session object which manages interactions with Amazon SageMaker APIs and any other AWS services needed. If not specified, creates one using the default AWS configuration chain. (Default: None).
schema_builder (SchemaBuilder, optional) – Schema builder for defining input/output schemas. If not specified, uses default schemas for the JumpStart model. (Default: None).

Returns:

A configured ModelBuilder instance ready to build and deploy: the specified JumpStart model.

Return type:

ModelBuilder

Example

>>> from sagemaker.core.jumpstart.configs import JumpStartConfig
>>> from sagemaker.serve.model_builder import ModelBuilder
>>>
>>> js_config = JumpStartConfig(
...     model_id="huggingface-llm-mistral-7b",
...     model_version="*"
... )
>>>
>>> from sagemaker.core.training.configs import Compute
>>>
>>> model_builder = ModelBuilder.from_jumpstart_config(
...     jumpstart_config=js_config,
...     compute=Compute(instance_type="ml.g5.2xlarge", instance_count=1)
... )
>>>
>>> model = model_builder.build()  # Creates Model resource
>>> endpoint = model_builder.deploy()  # Creates Endpoint resource
>>> result = endpoint.invoke(data=input_data)  # Make predictions

classmethod from_recommendation_job(recommendation_job, *, sagemaker_session: Session | None = None) → ModelBuilder[source]#

Create a ModelBuilder from an existing AIRecommendationJob.

Use this to deploy a recommendation from a job that this Python process didn’t run (e.g. a job kicked off in a notebook last week, or owned by a teammate). After construction, call deploy() with recommendation_index= or recommendation_spec_name= as you would after generate_deployment_recommendations.

Parameters:

recommendation_job – Either the name of an existing AIRecommendationJob or an AIRecommendationJob resource.
sagemaker_session – Optional SageMaker session. Used only when recommendation_job is a name and we have to .get() it.

Returns:

A ModelBuilder ready for .deploy(...).

Example

>>> mb = ModelBuilder.from_recommendation_job("my-rec-job")
>>> endpoint = mb.deploy(role=role, recommendation_index=0)

generate_deployment_recommendations(workload=None, performance_target: str | 'PerformanceTarget' | None = None, *, output_path: str | None = None, role_arn: str | None = None, instance_types: List[str] | None = None, capacity_reservation_arns: List[str] | None = None, advanced_optimization: bool = True, framework: str | 'InferenceFramework' | None = None, model_package_group: str | None = None, tags: Tags | None = None, job_name: str | None = None, workload_config_name: str | None = None, wait: bool = True, **workload_kwargs: Any)[source]#

Run an AIRecommendationJob for the model on this builder.

The service explores deployment configurations across the candidate instance types and ranks them against performance_target. Inspect ranked rows via self.recommendations; deploy a row with self.deploy(role=role) (top recommendation) or self.deploy(role=role, recommendation_index=N).

Parameters:

workload – Optional. A Workload instance, or the name/ARN of an existing AIWorkloadConfig. Omit and pass workload keyword arguments inline (tokenizer=, concurrency=, etc.) to construct a synthetic workload on the fly. tokenizer defaults to the model configured on this builder.
performance_target – Required. A PerformanceTarget member (THROUGHPUT, TTFT_MS, COST) or the equivalent string.
output_path – s3:// URI for recommendation output. Defaults to the session’s default bucket.
role_arn – IAM execution role ARN. Defaults to the role configured on this builder, then to the session’s execution role.
instance_types – Up to 3 candidate instance types to evaluate.
capacity_reservation_arns – Optional list of ML reservation ARNs.
advanced_optimization – If True (default), let the service explore optimization variants (speculative decoding, kernel tuning) on top of the candidate instance types.
framework – An InferenceFramework member (LMI, VLLM) or the equivalent string.
model_package_group – Optional model package group identifier.
tags – Optional resource tags.
job_name – Optional job name. Auto-generated if omitted.
workload_config_name – Optional name for the auto-created workload config. Auto-generated if omitted.
wait – If True (default), block until the job reaches a terminal state.
**workload_kwargs – Inline workload parameters, only used when workload is omitted. Forwarded to Workload.synthetic().

Returns:

The created AIRecommendationJob. When wait=True, the job is in a terminal state on return; self.recommendations is populated.

get_deployment_config() → Dict[str, Any] | None[source]#: Gets the deployment config to apply to the model.

image_config: Dict[str, str | PipelineVariable] | None = None#

image_uri: str | PipelineVariable | None = None#

inference_spec: InferenceSpec | None = None#

instance_type: str | None = None#

is_repack() → bool[source]#

Whether the source code needs to be repacked before uploading to S3.

Returns:: if the source need to be repacked or not
Return type:: bool

list_deployment_configs() → List[Dict[str, Any]][source]#: List deployment configs for the model in the current region.

log_level: int | None = 10#

mode: Mode | None = 3#

model: object | str | ModelTrainer | BaseTrainer | TrainingJob | ModelPackage | List[Model] | None = None#

model_metadata: Dict[str, Any] | None = None#

model_path: str | None#

model_server: ModelServer | None = None#

modelbuilder_list: List[ModelBuilder] | None = None#

network: Networking | None = None#

Create an optimized deployable Model instance with ModelBuilder.

Runs a SageMaker model optimization job to quantize, compile, or shard the model for improved inference performance. Returns a Model resource that can be deployed using the deploy() method.

Note: This returns a sagemaker.core.resources.Model object.

Parameters:

output_path (str, optional) – S3 URI where the optimized model artifacts will be stored. If not specified, uses the default output path. (Default: None).
instance_type (str, optional) – Target deployment instance type that the model is optimized for. For example, ‘ml.p4d.24xlarge’. (Default: None).
role_arn (str, optional) – IAM execution role ARN for the optimization job. If not specified, uses the role from initialization. (Default: None).
sagemaker_session (Session, optional) – Session object which manages interactions with Amazon SageMaker APIs and any other AWS services needed. If not specified, uses the session from initialization or creates one using the default AWS configuration chain. (Default: None).
region (str, optional) – The AWS region for the optimization job. If specified and different from the current region, a new session will be created. (Default: None).
model_name (str, optional) – The name for the optimized SageMaker model. If not specified, a unique name will be generated. (Default: None).
tags (Tags, optional) – Tags for labeling the model optimization job. (Default: None).
job_name (str, optional) – The name of the model optimization job. If not specified, a unique name will be generated. (Default: None).
accept_eula (bool, optional) – For models that require a Model Access Config, specify True or False to indicate whether model terms of use have been accepted. The accept_eula value must be explicitly defined as True in order to accept the end-user license agreement (EULA) that some models require. (Default: None).
quantization_config (Dict, optional) – Quantization configuration specifying the quantization method and parameters. For example: {‘OverrideEnvironment’: {‘OPTION_QUANTIZE’: ‘awq’}}. (Default: None).
compilation_config (Dict, optional) – Compilation configuration for optimizing the model for specific hardware. (Default: None).
speculative_decoding_config (Dict, optional) – Speculative decoding configuration for improving inference latency of large language models. (Default: None).
sharding_config (Dict, optional) – Model sharding configuration for distributing large models across multiple devices. (Default: None).
env_vars (Dict, optional) – Additional environment variables to pass to the optimization container. (Default: None).
vpc_config (Dict, optional) – VPC configuration for the optimization job. Should contain ‘Subnets’ and ‘SecurityGroupIds’ keys. (Default: None).
kms_key (str, optional) – KMS key ARN used to encrypt the optimized model artifacts when uploading to S3. (Default: None).
image_uri (str, optional) – Custom container image URI for the optimization job. If not specified, uses the default optimization container. (Default: None).
max_runtime_in_sec (int) – Maximum job execution time in seconds. The optimization job will be stopped if it exceeds this time. (Default: 36000).

Returns:

A sagemaker.core.resources.Model resource containing the optimized: model artifacts, ready for deployment.

Return type:

Model

Example

>>> model_builder = ModelBuilder(model=my_model, role_arn=role)
>>> optimized_model = model_builder.optimize(
...     instance_type="ml.g5.xlarge",
...     quantization_config={'OverrideEnvironment': {'OPTION_QUANTIZE': 'awq'}}
... )
>>> endpoint = model_builder.deploy()
>>> result = endpoint.invoke(data=input_data)

For deployment recommendations across candidate instance types and optimization variants, see generate_deployment_recommendations().

property recommendations#

Recommendation rows from the most recent generate_deployment_recommendations call.

Returns a list-like view; repr() renders a comparative table across rows, and .best is a shortcut for the top-ranked row. Each row forwards attribute access to the underlying service shape (e.g., rec.deployment_configuration.instance_type).

Deploy via self.deploy(recommendation_index=N) or self.deploy(recommendation_spec_name="...").

Empty if no recommendation job has been run, or if the job has not produced recommendations yet.

Creates a model package for creating SageMaker models or listing on Marketplace.

Parameters:

content_types (list[str] or list[PipelineVariable]) – The supported MIME types for the input data.
response_types (list[str] or list[PipelineVariable]) – The supported MIME types for the output data.
inference_instances (list[str] or list[PipelineVariable]) – A list of the instance types that are used to generate inferences in real-time (default: None).
transform_instances (list[str] or list[PipelineVariable]) – A list of the instance types on which a transformation job can be run or on which an endpoint can be deployed (default: None).
model_package_name (str or PipelineVariable) – Model Package name, exclusive to model_package_group_name, using model_package_name makes the Model Package un-versioned (default: None).
model_package_group_name (str or PipelineVariable) – Model Package Group name, exclusive to model_package_name, using model_package_group_name makes the Model Package versioned (default: None).
model_metrics (ModelMetrics) – ModelMetrics object (default: None).
metadata_properties (MetadataProperties) – MetadataProperties object (default: None).
marketplace_cert (bool) – A boolean value indicating if the Model Package is certified for AWS Marketplace (default: False).
approval_status (str or PipelineVariable) – Model Approval Status, values can be “Approved”, “Rejected”, or “PendingManualApproval” (default: “PendingManualApproval”).
description (str) – Model Package description (default: None).
drift_check_baselines (DriftCheckBaselines) – DriftCheckBaselines object (default: None).
customer_metadata_properties (dict[str, str] or dict[str, PipelineVariable]) – A dictionary of key-value paired metadata properties (default: None).
domain (str or PipelineVariable) – Domain values can be “COMPUTER_VISION”, “NATURAL_LANGUAGE_PROCESSING”, “MACHINE_LEARNING” (default: None).
task (str or PipelineVariable) – Task values which are supported by Inference Recommender are “FILL_MASK”, “IMAGE_CLASSIFICATION”, “OBJECT_DETECTION”, “TEXT_GENERATION”, “IMAGE_SEGMENTATION”, “CLASSIFICATION”, “REGRESSION”, “OTHER” (default: None).
sample_payload_url (str or PipelineVariable) – The S3 path where the sample payload is stored (default: None).
nearest_model_name (str or PipelineVariable) – Name of a pre-trained machine learning benchmarked by Amazon SageMaker Inference Recommender (default: None).
data_input_configuration (str or PipelineVariable) – Input object for the model (default: None).
skip_model_validation (str or PipelineVariable) – Indicates if you want to skip model validation. Values can be “All” or “None” (default: None).
source_uri (str or PipelineVariable) – The URI of the source for the model package (default: None).
model_card (ModeCard or ModelPackageModelCard) – document contains qualitative and quantitative information about a model (default: None).
model_life_cycle (ModelLifeCycle) – ModelLifeCycle object (default: None).
accept_eula (bool) – For models that require a Model Access Config, specify True or False to indicate whether model terms of use have been accepted (default: None).
model_type (JumpStartModelType) – Type of JumpStart model (default: None).

Returns:

A sagemaker.model.ModelPackage instance or pipeline step arguments in case the Model instance is built with PipelineSession

Note

The following parameters are inherited from ModelBuilder.__init__ and do not need to be passed to register(): - image_uri: Use self.image_uri (defined in __init__) - framework: Use self.framework (defined in __init__) - framework_version: Use self.framework_version (defined in __init__)

role_arn: str | None = None#

s3_model_data_url: str | PipelineVariable | Dict[str, Any] | None = None#

sagemaker_session: Session | None = None#

schema_builder: SchemaBuilder | None = None#

set_deployment_config(config_name: str, instance_type: str) → None[source]#: Sets the deployment config to apply to the model.

shared_libs: List[str]#

source_code: SourceCode | None = None#

to_string(obj: object)[source]#

Convert an object to string

This helper function handles converting PipelineVariable object to string as well

Parameters:: obj (object) – The object to be converted

transformer(instance_count, instance_type, strategy=None, assemble_with=None, output_path=None, output_kms_key=None, accept=None, env=None, max_concurrent_transforms=None, max_payload=None, tags=None, volume_kms_key=None)[source]#

Return a Transformer that uses this Model.

Parameters:

instance_count (int) – Number of EC2 instances to use.
instance_type (str) – Type of EC2 instance to use, for example, ‘ml.c4.xlarge’.
strategy (str) – The strategy used to decide how to batch records in a single request (default: None). Valid values: ‘MultiRecord’ and ‘SingleRecord’.
assemble_with (str) – How the output is assembled (default: None). Valid values: ‘Line’ or ‘None’.
output_path (str) – S3 location for saving the transform result. If not specified, results are stored to a default bucket.
output_kms_key (str) – Optional. KMS key ID for encrypting the transform output (default: None).
accept (str) – The accept header passed by the client to the inference endpoint. If it is supported by the endpoint, it will be the format of the batch transform output.
env (dict) – Environment variables to be set for use during the transform job (default: None).
max_concurrent_transforms (int) – The maximum number of HTTP requests to be made to each individual transform container at one time.
max_payload (int) – Maximum size of the payload in a single HTTP request to the container in MB.
tags (Optional[Tags]) – Tags for labeling a transform job. If none specified, then the tags used for the training job are used for the transform job.
volume_kms_key (str) – Optional. KMS key ID for encrypting the volume attached to the ML compute instance (default: None).

class sagemaker.serve.ModelServer(value)[source]#

Bases: Enum

An enum for model server

DJL_SERVING = 4#

LLAMACPP = 12#

MMS = 2#

SGLANG = 10#

SMD = 8#

TEI = 7#

TENSORFLOW_SERVING = 3#

TGI = 6#

TORCHSERVE = 1#

TRITON = 5#

VLLM = 9#

VLLM_OMNI = 11#

class sagemaker.serve.PerformanceTarget(value)[source]#

Bases: str, Enum

Optimization goal for a recommendation job.

COST = 'cost'#

THROUGHPUT = 'throughput'#

TTFT_MS = 'ttft-ms'#

class sagemaker.serve.RecommendationJob(*, ai_recommendation_job_name: str | PipelineVariable, ai_recommendation_job_arn: str | PipelineVariable | None = Unassigned(), ai_recommendation_job_status: str | PipelineVariable | None = Unassigned(), failure_reason: str | PipelineVariable | None = Unassigned(), model_source: AIModelSource | None = Unassigned(), output_config: AIRecommendationOutputResult | None = Unassigned(), inference_specification: AIRecommendationInferenceSpecification | None = Unassigned(), ai_workload_config_identifier: str | PipelineVariable | None = Unassigned(), optimize_model: bool | None = Unassigned(), performance_target: AIRecommendationPerformanceTarget | None = Unassigned(), recommendations: List[AIRecommendation] | None = Unassigned(), role_arn: str | PipelineVariable | None = Unassigned(), compute_spec: AIRecommendationComputeSpec | None = Unassigned(), creation_time: datetime | None = Unassigned(), start_time: datetime | None = Unassigned(), end_time: datetime | None = Unassigned(), tags: List[Tag] | None = Unassigned())[source]#

Bases: AIRecommendationJob

AIRecommendationJob with a one-shot result reader.

All standard lifecycle methods (refresh, wait, stop, delete) are inherited from the underlying resource; show_result is the only addition.

ai_recommendation_job_arn: str | PipelineVariable | None#

ai_recommendation_job_name: str | PipelineVariable#

ai_recommendation_job_status: str | PipelineVariable | None#

ai_workload_config_identifier: str | PipelineVariable | None#

compute_spec: AIRecommendationComputeSpec | None#

creation_time: datetime | None#

end_time: datetime | None#

failure_reason: str | PipelineVariable | None#

inference_specification: AIRecommendationInferenceSpecification | None#

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'protected_namespaces': (), 'validate_assignment': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_source: AIModelSource | None#

optimize_model: bool | None#

output_config: AIRecommendationOutputResult | None#

performance_target: AIRecommendationPerformanceTarget | None#

recommendations: List[AIRecommendation] | None#

role_arn: str | PipelineVariable | None#

show_result() → _RecommendationsView[source]#

Return the ranked recommendations produced by the job.

Returns the same list-like view as ModelBuilder.recommendations: repr() renders a comparative table across rows, .best is the top-ranked row, and each row pretty-prints and forwards attribute access to the underlying service shape. The job must be in a terminal state; show_result calls refresh() once but does not poll.

start_time: datetime | None#

tags: List[Tag] | None#

class sagemaker.serve.Secret(arn: str, _created: bool = False, _session: Any | None = None)[source]#

Bases: object

A handle to a Secrets Manager secret.

Supports context-manager use. On context exit, only a secret this object created (via from_string()) is deleted; a secret you merely wrapped by ARN is left untouched.

arn: str#

delete(*, force_delete_without_recovery: bool = False, session: boto3.session.Session | None = None) → None[source]#

Delete the underlying Secrets Manager secret.

Parameters:

force_delete_without_recovery – If True, delete immediately with no recovery window. Defaults to False, which uses Secrets Manager’s recovery window so an accidental delete can be restored.
session – Optional boto3 session. Defaults to the session that created this secret, then to the ambient session.

classmethod from_string(value: str, *, name: str | None = None, session: boto3.session.Session | None = None) → Secret[source]#

Create a new Secrets Manager secret from a plaintext value.

Note: this helper creates the secret with default permissions only. If the consuming workload needs to read the secret at job runtime, you may need to attach a resource policy granting secretsmanager:GetSecretValue to the appropriate principal.

Parameters:

value – The plaintext secret value to store.
name – Optional secret name. Defaults to sagemaker-workload-<uuid>.
session – Optional boto3 session. Defaults to the ambient session.

Returns:

A Secret referencing the newly created secret.

class sagemaker.serve.Workload(*, parameters: ~typing.Dict[str, ~typing.Any], secrets: ~typing.Dict[str, str | ~sagemaker.serve.ai_inference_recommender.secrets.Secret] = <factory>, tooling: ~typing.Dict[str, ~typing.Any] = <factory>, dataset_channels: ~typing.List[~sagemaker.serve.ai_inference_recommender.workload._DatasetChannel] = <factory>)[source]#

Bases: BaseModel

A workload specification used by benchmark and recommendation jobs.

dataset_channels: List[_DatasetChannel]#

classmethod from_dataset(s3_uri: str, *, custom_dataset_type: str | None = None, tokenizer: str | None = None, concurrency: int = 1, request_count: int = 100, prompt_input_tokens_mean: int = 256, prompt_input_tokens_stddev: float = 0.0, output_tokens_mean: int = 256, output_tokens_stddev: float = 0.0, streaming: bool = True, hf_token: str | Secret | None = None, channel_name: str = 'dataset', **params: Any) → Workload[source]#

Build a workload that drives traffic from an S3-hosted dataset.

The benchmark replays requests from the dataset at the given S3 prefix.

Parameters:

s3_uri – s3://bucket/prefix/ URI containing the dataset.
custom_dataset_type – Optional AIPerf custom-dataset format (e.g. "openai-chat").
tokenizer – Optional HuggingFace tokenizer id; required for some AIPerf metrics that compute per-token statistics.
concurrency – Number of in-flight requests.
request_count – Total number of requests to issue.
prompt_input_tokens_mean – Mean input prompt length in tokens.
prompt_input_tokens_stddev – Standard deviation of input token count.
output_tokens_mean – Mean output response length in tokens.
output_tokens_stddev – Standard deviation of output token count.
streaming – Whether to use streaming chat completions.
hf_token – HuggingFace access token for gated tokenizers. Accepts a Secret or a Secrets Manager ARN string.
channel_name – Name of the input data channel the dataset is mounted under. Defaults to "dataset".
**params – Additional AIPerf parameters for the workload.

Returns:

A Workload configured to drive traffic from the dataset.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parameters: Dict[str, Any]#

secrets: Dict[str, str | Secret]#

classmethod sonnet(**kwargs: Any) → Workload[source]#

Alias for synthetic().

AIPerf seeds synthetic prompts from the Sonnet dataset by default, so Workload.sonnet(...) is the same as Workload.synthetic(...).

classmethod synthetic(*, tokenizer: str, concurrency: int = 1, request_count: int = 100, prompt_input_tokens_mean: int = 256, prompt_input_tokens_stddev: float = 0.0, output_tokens_mean: int = 256, output_tokens_stddev: float = 0.0, streaming: bool = True, hf_token: str | Secret | None = None, **params: Any) → Workload[source]#

Build a workload that uses synthetic prompts.

Synthetic prompts are generated by AIPerf from the Sonnet dataset, producing realistic token distributions. Use Workload.from_dataset() to drive the benchmark from a real request trace instead.

Parameters:

tokenizer – HuggingFace tokenizer id (e.g. meta-llama/Llama-3.2-1B).
concurrency – Number of in-flight requests.
request_count – Total number of requests to issue.
prompt_input_tokens_mean – Mean input prompt length in tokens.
prompt_input_tokens_stddev – Standard deviation of input token count.
output_tokens_mean – Mean output response length in tokens.
output_tokens_stddev – Standard deviation of output token count.
streaming – Whether to use streaming chat completions.
hf_token – HuggingFace access token for gated tokenizers. Accepts a Secret or a Secrets Manager ARN string.
**params – Additional parameters merged into the workload’s parameters map.

classmethod template(request_template: str | None = None, *, template_s3_uri: str | None = None, sagemaker_session: Any | None = None, response_field: str | None = None, tokenizer: str | None = None, concurrency: int = 1, request_count: int = 100, prompt_input_tokens_mean: int = 256, prompt_input_tokens_stddev: float = 0.0, output_tokens_mean: int = 256, output_tokens_stddev: float = 0.0, streaming: bool = True, hf_token: str | Secret | None = None, channel_name: str = 'template', extra_inputs: str | None = None, **params: Any) → Workload[source]#

Build a workload that benchmarks a custom-format (non-OpenAI) endpoint.

Use this for endpoints that don’t speak the OpenAI chat-completions format (e.g. DJL custom handlers, TensorRT-LLM native format). You provide a Jinja2 template describing your endpoint’s request payload; it is rendered per request (still using synthetic prompts) and sent to your endpoint. A JMESPath response_field extracts the generated text from the response.

Provide the template in one of two ways: pass request_template as a local file path or an inline Jinja2 string and it is uploaded to the session default bucket for you, or pass template_s3_uri if the template is already in S3. Exactly one of the two is required.

Parameters:

request_template – The Jinja2 template for your endpoint’s request payload, given as either a local file path or an inline string. It is uploaded to the session default bucket for you; mutually exclusive with template_s3_uri.
template_s3_uri – s3://bucket/key URI of an already-uploaded Jinja2 template file (a single object, not a prefix). Use this only when the template is already in S3; otherwise pass request_template.
sagemaker_session – Session used to upload a local/inline request_template. Defaults to a new Session.
response_field – Optional JMESPath query that extracts the response text from your endpoint’s output (e.g. "generated_text" or "choices[0].message.content"). Omit to let AIPerf auto-detect the response format.
tokenizer – Optional HuggingFace tokenizer id; required for AIPerf metrics that compute per-token statistics.
concurrency – Number of in-flight requests.
request_count – Total number of requests to issue.
prompt_input_tokens_mean – Mean input prompt length in tokens.
prompt_input_tokens_stddev – Standard deviation of input token count.
output_tokens_mean – Mean output response length in tokens.
output_tokens_stddev – Standard deviation of output token count.
streaming – Whether the endpoint streams its response.
hf_token – HuggingFace access token for gated tokenizers. Accepts a Secret or a Secrets Manager ARN string.
channel_name – Name of the input data channel the template is mounted under. Defaults to "template".
extra_inputs – Optional space-separated key:value pairs for advanced AIPerf options beyond response_field (e.g. "ignore_eos:true").
**params – Additional AIPerf parameters for the workload.

Returns:

A Workload configured for template mode.

to_inline() → str[source]#

Serialize the workload to a JSON string.

Secret values are flattened to their ARN strings.

tooling: Dict[str, Any]#

exception sagemaker.serve.WorkloadValidationError(message='', **kwargs)[source]#

Bases: ValidationError

Raised when the server rejects a workload spec.

fmt = 'Server rejected workload: {message}'#

Start an AI benchmark job against a SageMaker endpoint.

Parameters:

endpoint – An Endpoint resource, or the name/ARN of an existing endpoint to benchmark.
workload – Optional. A Workload instance, or the name/ARN of an existing AIWorkloadConfig. Omit this and pass workload keyword arguments inline (tokenizer=, concurrency=, etc.) to construct a synthetic workload on the fly.
output_path – s3:// URI for benchmark output. Defaults to the session’s default bucket.
role – IAM execution role ARN. Defaults to the SageMaker execution role from the ambient session.
inference_components – Optional list of inference component names to target on the endpoint.
vpc_config – Optional VpcConfig for VPC-only endpoints.
tags – Optional resource tags.
name – Optional benchmark job name. Auto-generated if omitted.
workload_config_name – Optional name for the auto-created workload config. Auto-generated if omitted.
sagemaker_session – Session used to create the benchmark job and workload config. Defaults to the passed Endpoint’s session, then to a new Session.
wait – If True (default), block until the job reaches a terminal state.
**workload_kwargs – Inline workload parameters. Only used when workload is omitted; forwarded to Workload.synthetic.

Returns:

The created BenchmarkJob. Once terminal, call job.show_result() to download and parse the metrics.

SageMaker Serve

Contents

SageMaker Serve#

Model Deployment#