sagemaker.core.jumpstart.utils

sagemaker.core.jumpstart.utils#

This module contains utils for JumpStart.

Functions

`add_bedrock_store_tags`(tags, compatibility)	Adds custom Hub arn tag to JumpStart related resources.
`add_hub_content_arn_tags`(tags, hub_content_arn)	Adds custom Hub arn tag to JumpStart related resources.
`add_instance_rate_stats_to_benchmark_metrics`(...)	Adds instance types metric stats to the given benchmark_metrics dict.
`add_jumpstart_model_info_tags`(tags, ...[, ...])	Add custom model ID and version tags to JumpStart related resources.
`add_jumpstart_uri_tags`([tags, ...])	Add custom uri tags to JumpStart models, return the updated tags.
`add_single_jumpstart_tag`(tag_value, tag_key, ...)	Adds `tag_key` to `curr_tags` if `uri` corresponds to a JumpStart model.
`deployment_config_response_data`(...)	Deployment config api response data.
`emit_logs_based_on_model_specs`(model_specs, ...)	Emits logs based on model specs and region.
`get_benchmark_stats`(region, model_id, ...[, ...])	Returns benchmark stats for the given model ID and region.
`get_config_names`(region, model_id, model_version)	Returns a list of config names for the given model ID and region.
`get_default_jumpstart_session_with_user_agent_suffix`([...])	Returns default JumpStart SageMaker Session with model-specific user agent suffix.
`get_draft_model_content_bucket`(provider, region)	Returns the correct content bucket for a 1p draft model.
`get_eula_message`(model_specs, region)	Returns EULA message to display if one is available, else empty string.
`get_eula_url`(document[, sagemaker_session])	Get the EULA URL from the HubContentDocument.
`get_formatted_eula_message_template`(...)	Returns a formatted EULA message.
`get_formatted_manifest`(manifest)	Returns formatted manifest dictionary from raw manifest.
`get_hub_access_config`(hub_content_arn)	Get hub access config
`get_jumpstart_base_name_if_jumpstart_model`(*uris)	Return default JumpStart base name if a URI belongs to JumpStart.
`get_jumpstart_configs`(region, model_id, ...)	Returns metadata configs for the given model ID and region.
`get_jumpstart_content_bucket`([region])	Returns the regionalized content bucket name for JumpStart.
`get_jumpstart_gated_content_bucket`([region])	Returns regionalized private content bucket name for JumpStart.
`get_jumpstart_launched_regions_message`()	Returns formatted string indicating where JumpStart is launched.
`get_jumpstart_model_info_from_resource_arn`(...)	Returns the JumpStart model ID, version and config name if in resource tags.
`get_jumpstart_user_agent_extra_suffix`(...)	Returns the model-specific user agent string to be added to requests.
`get_latest_version`(versions)	Returns the latest version using sem-ver when possible.
`get_metrics_from_deployment_configs`(...)	Extracts benchmark metrics from deployment configs metadata.
`get_model_access_config`(accept_eula)	Get access configs
`get_neo_content_bucket`([region])	Returns the regionalized S3 bucket name for Neo service.
`get_region_fallback`([s3_bucket_name, ...])	Returns region to use for JumpStart functionality implicitly via session objects.
`get_sagemaker_version`()	Returns sagemaker library version.
`get_tag_value`(tag_key, tag_array)	Return the value of a tag whose key matches the given `tag_key`.
`get_top_ranked_config_name`(region, model_id, ...)	Returns the top ranked config name for the given model ID and region.
`has_instance_rate_stat`(benchmark_metric_stats)	Determines whether a benchmark metric stats contains instance rate metric stat.
`is_jumpstart_model_input`(model_id, version)	Determines if model_id and version input are for JumpStart.
`is_jumpstart_model_uri`(uri)	Returns True if URI corresponds to a JumpStart-hosted model.
`is_pipeline_variable`(var)	Check if the variable is a pipeline variable
`parse_sagemaker_version`()	Returns sagemaker library version.
`remove_env_var_from_estimator_kwargs_if_accept_eula_present`(...)	Remove env vars if access configs are used
`resolve_estimator_sagemaker_config_field`(...)	Given a field name, checks if there is a sagemaker config value to set.
`resolve_model_sagemaker_config_field`(...[, ...])	Given a field name, checks if there is a sagemaker config value to set.
`tag_key_in_array`(tag_key, tag_array)	Returns True if `tag_key` is in the `tag_array`.
`update_dict_if_key_not_present`(...)	If a key is not present in the dict, add the new (key, value) pair, and return dict.
`update_inference_tags_with_jumpstart_training_tags`(...)	Updates the tags for the `sagemaker.model.Model.deploy` command with any JumpStart tags.
`validate_model_id_and_get_type`(model_id[, ...])	Returns model type if the model ID is supported for the given script.
`verify_model_region_and_return_specs`(...[, ...])	Verifies that an acceptable model_id, version, scope, and region combination is provided.

sagemaker.core.jumpstart.utils.add_bedrock_store_tags(tags: List[Dict[str, str | PipelineVariable]] | None, compatibility: str) → List[Dict[str, str | PipelineVariable]] | None[source]#: Adds custom Hub arn tag to JumpStart related resources.

sagemaker.core.jumpstart.utils.add_hub_content_arn_tags(tags: List[Dict[str, str | PipelineVariable]] | None, hub_content_arn: str) → List[Dict[str, str | PipelineVariable]] | None[source]#: Adds custom Hub arn tag to JumpStart related resources.

sagemaker.core.jumpstart.utils.add_instance_rate_stats_to_benchmark_metrics(region: str, benchmark_metrics: Dict[str, List[JumpStartBenchmarkStat]] | None) → Tuple[Dict[str, str], Dict[str, List[JumpStartBenchmarkStat]]] | None[source]#

Adds instance types metric stats to the given benchmark_metrics dict.

Parameters:

region (str) – AWS region.
benchmark_metrics (Optional[Dict[str, List[JumpStartBenchmarkStat]]])

Returns:

Contains Error and metrics.

Return type:

Optional[Tuple[Dict[str, str], Dict[str, List[JumpStartBenchmarkStat]]]]

sagemaker.core.jumpstart.utils.add_jumpstart_model_info_tags(tags: List[Dict[str, str | PipelineVariable]] | None, model_id: str, model_version: str, model_type: JumpStartModelType | None = None, config_name: str | None = None, scope: JumpStartScriptScope | None = None) → List[Dict[str, str | PipelineVariable]][source]#: Add custom model ID and version tags to JumpStart related resources.

Add custom uri tags to JumpStart models, return the updated tags.

No-op if this is not a JumpStart model related resource.

Parameters:

tags (Optional[List[Dict[str,str]]) – Current tags for JumpStart inference or training job. (Default: None).
inference_model_uri (Optional[Union[dict, str]]) – S3 URI for inference model artifact. (Default: None).
inference_script_uri (Optional[str]) – S3 URI for inference script tarball. (Default: None).
training_model_uri (Optional[str]) – S3 URI for training model artifact. (Default: None).
training_script_uri (Optional[str]) – S3 URI for training script tarball. (Default: None).

sagemaker.core.jumpstart.utils.add_single_jumpstart_tag(tag_value: str, tag_key: JumpStartTag, curr_tags: List[Dict[str, str]] | None, is_uri=False) → List | None[source]#

Adds tag_key to curr_tags if uri corresponds to a JumpStart model.

Parameters:

uri (str) – URI which may correspond to a JumpStart model.
tag_key (enums.JumpStartTag) – Custom tag to apply to current tags if the URI corresponds to a JumpStart model.
curr_tags (Optional[List]) – Current tags associated with Estimator or Model.
is_uri (boolean) – Set to True to indicate a s3 uri is to be tagged. Set to False to indicate tags for JumpStart model id / version are being added. (Default: False).

sagemaker.core.jumpstart.utils.deployment_config_response_data(deployment_configs: List[DeploymentConfigMetadata] | None) → List[Dict[str, Any]][source]#

Deployment config api response data.

Parameters:

deployment_configs (Optional[List[DeploymentConfigMetadata]])
metadata. (List of deployment configs)

Returns:

List of deployment config api response data.

Return type:

List[Dict[str, Any]]

sagemaker.core.jumpstart.utils.emit_logs_based_on_model_specs(model_specs: JumpStartModelSpecs, region: str, s3_client: client) → None[source]#: Emits logs based on model specs and region.

sagemaker.core.jumpstart.utils.get_benchmark_stats(region: str, model_id: str, model_version: str, config_names: ~typing.List[str] | None = None, hub_arn: str | None = None, sagemaker_session: ~sagemaker.core.helper.session_helper.Session | None = <sagemaker.core.helper.session_helper.Session object>, scope: ~sagemaker.core.jumpstart.enums.JumpStartScriptScope = JumpStartScriptScope.INFERENCE, model_type: ~sagemaker.core.jumpstart.enums.JumpStartModelType = JumpStartModelType.OPEN_WEIGHTS) → Dict[str, List[JumpStartBenchmarkStat]][source]#

Returns benchmark stats for the given model ID and region.

Raises:: ValueError – If the script scope is not supported by JumpStart.

sagemaker.core.jumpstart.utils.get_config_names(region: str, model_id: str, model_version: str, sagemaker_session: ~sagemaker.core.helper.session_helper.Session | None = <sagemaker.core.helper.session_helper.Session object>, scope: ~sagemaker.core.jumpstart.enums.JumpStartScriptScope = JumpStartScriptScope.INFERENCE, model_type: ~sagemaker.core.jumpstart.enums.JumpStartModelType = JumpStartModelType.OPEN_WEIGHTS) → List[str][source]#

Returns a list of config names for the given model ID and region.

Raises:: ValueError – If the script scope is not supported by JumpStart.

sagemaker.core.jumpstart.utils.get_default_jumpstart_session_with_user_agent_suffix(model_id: str | None = None, model_version: str | None = None, config_name: str | None = None, is_hub_content: bool | None = False) → Session[source]#: Returns default JumpStart SageMaker Session with model-specific user agent suffix.

sagemaker.core.jumpstart.utils.get_draft_model_content_bucket(provider: Dict, region: str) → str[source]#: Returns the correct content bucket for a 1p draft model.

sagemaker.core.jumpstart.utils.get_eula_message(model_specs: JumpStartModelSpecs, region: str) → str[source]#: Returns EULA message to display if one is available, else empty string.

sagemaker.core.jumpstart.utils.get_eula_url(document: HubContentDocument, sagemaker_session: Session | None = None) → str[source]#

Get the EULA URL from the HubContentDocument.

Parameters:

document (HubContentDocument) – The HubContentDocument object.
sagemaker_session (Optional[Session]) – SageMaker session. Defaults to None.

Returns:

The EULA URL.

Return type:

str

sagemaker.core.jumpstart.utils.get_formatted_eula_message_template(model_id: str, region: str, hosting_eula_key: str) → str[source]#: Returns a formatted EULA message.

sagemaker.core.jumpstart.utils.get_formatted_manifest(manifest: List[Dict]) → Dict[JumpStartVersionedModelId, JumpStartModelHeader][source]#

Returns formatted manifest dictionary from raw manifest.

Keys are JumpStartVersionedModelId objects, values are JumpStartModelHeader objects.

sagemaker.core.jumpstart.utils.get_hub_access_config(hub_content_arn: str | None)[source]#

Get hub access config

Parameters:: hub_content_arn (Optional[bool]) – Arn of the model reference hub content

sagemaker.core.jumpstart.utils.get_jumpstart_base_name_if_jumpstart_model(*uris: str | None) → str | None[source]#

Return default JumpStart base name if a URI belongs to JumpStart.

If no URIs belong to JumpStart, return None.

Parameters:: *uris (Optional[str]) – URI to test for association with JumpStart.

sagemaker.core.jumpstart.utils.get_jumpstart_configs(region: str, model_id: str, model_version: str, config_names: ~typing.List[str] | None = None, sagemaker_session: ~sagemaker.core.helper.session_helper.Session | None = <sagemaker.core.helper.session_helper.Session object>, scope: ~sagemaker.core.jumpstart.enums.JumpStartScriptScope = JumpStartScriptScope.INFERENCE, model_type: ~sagemaker.core.jumpstart.enums.JumpStartModelType = JumpStartModelType.OPEN_WEIGHTS, hub_arn: str | None = None) → Dict[str, JumpStartMetadataConfig][source]#

Returns metadata configs for the given model ID and region.

Raises:: ValueError – If the script scope is not supported by JumpStart.

sagemaker.core.jumpstart.utils.get_jumpstart_content_bucket(region: str = 'us-west-2') → str[source]#

Returns the regionalized content bucket name for JumpStart.

Raises:: ValueError – If JumpStart is not launched in region.

sagemaker.core.jumpstart.utils.get_jumpstart_gated_content_bucket(region: str = 'us-west-2') → str[source]#

Returns regionalized private content bucket name for JumpStart.

Raises:: ValueError – If JumpStart is not launched in region or private content unavailable in that region.

sagemaker.core.jumpstart.utils.get_jumpstart_launched_regions_message() → str[source]#: Returns formatted string indicating where JumpStart is launched.

sagemaker.core.jumpstart.utils.get_jumpstart_model_info_from_resource_arn(resource_arn: str, sagemaker_session: ~sagemaker.core.helper.session_helper.Session = <sagemaker.core.helper.session_helper.Session object>) → Tuple[str | None, str | None, str | None, str | None][source]#

Returns the JumpStart model ID, version and config name if in resource tags.

Returns ‘None’ if model ID or version or config name cannot be inferred from tags.

sagemaker.core.jumpstart.utils.get_jumpstart_user_agent_extra_suffix(model_id: str | None, model_version: str | None, config_name: str | None, is_hub_content: bool | None) → str[source]#: Returns the model-specific user agent string to be added to requests.

sagemaker.core.jumpstart.utils.get_latest_version(versions: List[str]) → str | None[source]#: Returns the latest version using sem-ver when possible.

sagemaker.core.jumpstart.utils.get_metrics_from_deployment_configs(deployment_configs: List[DeploymentConfigMetadata] | None) → Dict[str, List[str]][source]#

Extracts benchmark metrics from deployment configs metadata.

Parameters:

deployment_configs (Optional[List[DeploymentConfigMetadata]])
metadata. (List of deployment configs)

Returns:

Deployment configs bench metrics dict.

Return type:

Dict[str, List[str]]

sagemaker.core.jumpstart.utils.get_model_access_config(accept_eula: bool | None)[source]#

Get access configs

Parameters:: accept_eula (Optional[bool]) – Whether or not the EULA was accepted, optionally passed in to Estimator.fit().

sagemaker.core.jumpstart.utils.get_neo_content_bucket(region: str = 'us-west-2') → str[source]#

Returns the regionalized S3 bucket name for Neo service.

Raises:: ValueError – If Neo is not launched in region.

sagemaker.core.jumpstart.utils.get_region_fallback(s3_bucket_name: str | None = None, s3_client: client | None = None, sagemaker_session: Session | None = None) → str[source]#: Returns region to use for JumpStart functionality implicitly via session objects.

sagemaker.core.jumpstart.utils.get_sagemaker_version() → str[source]#

Returns sagemaker library version.

If the sagemaker library version has not been set, this function calls parse_sagemaker_version to retrieve the version and set the constant.

sagemaker.core.jumpstart.utils.get_tag_value(tag_key: str, tag_array: List[Dict[str, str]]) → str[source]#

Return the value of a tag whose key matches the given tag_key.

Parameters:

tag_key (str) – AWS tag for which to search.
tag_array (List[Dict[str, str]]) – List of AWS tags, each formatted as dicts.

Raises:

KeyError – If the number of matches for the tag_key is not equal to 1.

sagemaker.core.jumpstart.utils.get_top_ranked_config_name(region: str, model_id: str, model_version: str, sagemaker_session: ~sagemaker.core.helper.session_helper.Session | None = <sagemaker.core.helper.session_helper.Session object>, scope: ~sagemaker.core.jumpstart.enums.JumpStartScriptScope = JumpStartScriptScope.INFERENCE, model_type: ~sagemaker.core.jumpstart.enums.JumpStartModelType = JumpStartModelType.OPEN_WEIGHTS, tolerate_deprecated_model: bool = False, tolerate_vulnerable_model: bool = False, hub_arn: str | None = None, ranking_name: ~sagemaker.core.jumpstart.enums.JumpStartConfigRankingName = JumpStartConfigRankingName.DEFAULT) → str | None[source]#

Returns the top ranked config name for the given model ID and region.

Raises:: ValueError – If the script scope is not supported by JumpStart.

sagemaker.core.jumpstart.utils.has_instance_rate_stat(benchmark_metric_stats: List[JumpStartBenchmarkStat] | None) → bool[source]#

Determines whether a benchmark metric stats contains instance rate metric stat.

Parameters:

benchmark_metric_stats (Optional[List[JumpStartBenchmarkStat]])
stats. (List of benchmark metric)

Returns:

Whether the benchmark metric stats contains instance rate metric stat.

Return type:

bool

sagemaker.core.jumpstart.utils.is_jumpstart_model_input(model_id: str | None, version: str | None) → bool[source]#

Determines if model_id and version input are for JumpStart.

This method returns True if both arguments are not None, false if both arguments are None, and raises an exception if one argument is None but the other isn’t.

Parameters:

model_id (str) – Optional. Model ID of the JumpStart model.
version (str) – Optional. Version of the JumpStart model.

Raises:

ValueError – If only one of the two arguments is None.

sagemaker.core.jumpstart.utils.is_jumpstart_model_uri(uri: str | None) → bool[source]#

Returns True if URI corresponds to a JumpStart-hosted model.

Parameters:: uri (Optional[str]) – uri for inference/training job.

sagemaker.core.jumpstart.utils.is_pipeline_variable(var: object) → bool[source]#: Check if the variable is a pipeline variable

sagemaker.core.jumpstart.utils.parse_sagemaker_version() → str[source]#

Returns sagemaker library version. This should only be called once.

Function reads __version__ variable in sagemaker module. In order to maintain compatibility with the packaging.version library, versions with fewer than 2, or more than 3, periods are rejected. All versions that cannot be parsed with packaging.version are also rejected.

Raises:

RuntimeError – If the SageMaker version is not readable. An exception is also raised if
the version cannot be parsed by packaging.version. –

sagemaker.core.jumpstart.utils.remove_env_var_from_estimator_kwargs_if_accept_eula_present(init_kwargs: dict, accept_eula: bool | None)[source]#

Remove env vars if access configs are used

Parameters:

init_kwargs (dict) – Dictionary of kwargs when Estimator is instantiated.
accept_eula (Optional[bool]) – Whether or not the EULA was accepted, optionally passed in to Estimator.fit().

sagemaker.core.jumpstart.utils.resolve_estimator_sagemaker_config_field(field_name: str, field_val: Any | None, sagemaker_session: Session, default_value: str | None = None) → Any[source]#

Given a field name, checks if there is a sagemaker config value to set.

For the role field, which is customer-supplied, we allow field_val to take precedence over sagemaker config values. For all other fields, sagemaker config values take precedence over the JumpStart default fields.

sagemaker.core.jumpstart.utils.resolve_model_sagemaker_config_field(field_name: str, field_val: Any | None, sagemaker_session: Session, default_value: str | None = None) → Any[source]#

Given a field name, checks if there is a sagemaker config value to set.

For the role field, which is customer-supplied, we allow field_val to take precedence over sagemaker config values. For all other fields, sagemaker config values take precedence over the JumpStart default fields.

sagemaker.core.jumpstart.utils.tag_key_in_array(tag_key: str, tag_array: List[Dict[str, str]]) → bool[source]#

Returns True if tag_key is in the tag_array.

Parameters:

tag_key (str) – the tag key to check if it’s already in the tag_array.
tag_array (List[Dict[str, str]]) – array of tags to check for tag_key.

sagemaker.core.jumpstart.utils.update_dict_if_key_not_present(dict_to_update: dict | None, key_to_add: Any, value_to_add: Any) → dict | None[source]#

If a key is not present in the dict, add the new (key, value) pair, and return dict.

If dict is empty, return None.

sagemaker.core.jumpstart.utils.update_inference_tags_with_jumpstart_training_tags(inference_tags: List[Dict[str, str]] | None, training_tags: List[Dict[str, str]] | None) → List[Dict[str, str]] | None[source]#

Updates the tags for the sagemaker.model.Model.deploy command with any JumpStart tags.

Parameters:

inference_tags (Optional[List[Dict[str, str]]]) – Custom tags to appy to inference job.
training_tags (Optional[List[Dict[str, str]]]) – Tags from training job.

sagemaker.core.jumpstart.utils.validate_model_id_and_get_type(model_id: str | None, region: str | None = None, model_version: str | None = None, script: ~sagemaker.core.jumpstart.enums.JumpStartScriptScope = JumpStartScriptScope.INFERENCE, sagemaker_session: ~sagemaker.core.helper.session_helper.Session | None = <sagemaker.core.helper.session_helper.Session object>, hub_arn: str | None = None) → JumpStartModelType | None[source]#

Returns model type if the model ID is supported for the given script.

Raises:: ValueError – If the script is not supported by JumpStart.

sagemaker.core.jumpstart.utils.verify_model_region_and_return_specs(model_id: str | None, version: str | None, scope: str | None, region: str | None = None, hub_arn: str | None = None, tolerate_vulnerable_model: bool = False, tolerate_deprecated_model: bool = False, sagemaker_session: ~sagemaker.core.helper.session_helper.Session = <sagemaker.core.helper.session_helper.Session object>, model_type: ~sagemaker.core.jumpstart.enums.JumpStartModelType = JumpStartModelType.OPEN_WEIGHTS, config_name: str | None = None) → JumpStartModelSpecs[source]#

Verifies that an acceptable model_id, version, scope, and region combination is provided.

Parameters:

model_id (Optional[str]) – model ID of the JumpStart model to verify and obtains specs.
version (Optional[str]) – version of the JumpStart model to verify and obtains specs.
scope (Optional[str]) – scope of the JumpStart model to verify.
region (Optional[str]) – region of the JumpStart model to verify and obtains specs.
hub_arn (str) – The arn of the SageMaker Hub for which to retrieve model details from. (default: None).
tolerate_vulnerable_model (bool) – True if vulnerable versions of model specifications should be tolerated (exception not raised). If False, raises an exception if the script used by this version of the model has dependencies with known security vulnerabilities. (Default: False).
tolerate_deprecated_model (bool) – True if deprecated models should be tolerated (exception not raised). False if these models should raise an exception. (Default: False).
sagemaker_session (sagemaker.session.Session) – A SageMaker Session object, used for SageMaker interactions. If not specified, one is created using the default AWS configuration chain. (Default: sagemaker.jumpstart.constants.DEFAULT_JUMPSTART_SAGEMAKER_SESSION).
config_name (Optional[str]) – Name of the JumpStart Model config to apply. (Default: None).

Raises:

NotImplementedError – If the scope is not supported.
ValueError – If the combination of arguments specified is not supported.
VulnerableJumpStartModelError – If any of the dependencies required by the script have known security vulnerabilities.
DeprecatedJumpStartModelError – If the version of the model is deprecated.

sagemaker.core.jumpstart.utils

Contents

sagemaker.core.jumpstart.utils#