sagemaker.serve.ai_inference_recommender

SageMaker GenAI inference benchmarking and recommendation.

class sagemaker.serve.ai_inference_recommender.BenchmarkJob(*, ai_benchmark_job_name: str | PipelineVariable, ai_benchmark_job_arn: str | PipelineVariable | None = Unassigned(), ai_benchmark_job_status: str | PipelineVariable | None = Unassigned(), failure_reason: str | PipelineVariable | None = Unassigned(), benchmark_target: AIBenchmarkTarget | None = Unassigned(), output_config: AIBenchmarkOutputResult | None = Unassigned(), ai_workload_config_identifier: str | PipelineVariable | None = Unassigned(), role_arn: str | PipelineVariable | None = Unassigned(), network_config: AIBenchmarkNetworkConfig | None = Unassigned(), creation_time: datetime | None = Unassigned(), start_time: datetime | None = Unassigned(), end_time: datetime | None = Unassigned(), tags: List[Tag] | None = Unassigned())

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

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.ai_inference_recommender.BenchmarkMetric(name: str, unit: str | None = None, avg: float | None = None, min: float | None = None, max: float | None = None, p50: float | None = None, p90: float | None = None, p95: float | None = None, p99: float | None = None, stddev: float | None = None, raw: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

A single benchmark metric with its statistical aggregates.

avg: float | None = None

classmethod from_dict(name: str, data: Dict[str, Any]) → BenchmarkMetric

max: float | None = None

min: float | None = None

name: str

p50: float | None = None

p90: float | None = None

p95: float | None = None

p99: float | None = None

raw: Dict[str, Any]

stddev: float | None = None

unit: str | None = None

class sagemaker.serve.ai_inference_recommender.BenchmarkMetrics(request_throughput: ~sagemaker.serve.ai_inference_recommender.result.BenchmarkMetric | None = None, request_latency: ~sagemaker.serve.ai_inference_recommender.result.BenchmarkMetric | None = None, time_to_first_token: ~sagemaker.serve.ai_inference_recommender.result.BenchmarkMetric | None = None, inter_token_latency: ~sagemaker.serve.ai_inference_recommender.result.BenchmarkMetric | None = None, output_token_throughput: ~sagemaker.serve.ai_inference_recommender.result.BenchmarkMetric | None = None, all_metrics: ~typing.Dict[str, ~sagemaker.serve.ai_inference_recommender.result.BenchmarkMetric] = <factory>)

Bases: object

Typed access to the well-known AIPerf metrics.

Use .get(name) to look up any metric by its raw key. print()-ing this object renders every metric in a table; print(result) (the parent BenchmarkResult) shows just the well-known metrics.

all_metrics: Dict[str, BenchmarkMetric]

classmethod from_profile_json(profile: Dict[str, Any]) → BenchmarkMetrics

get(name: str) → BenchmarkMetric | None

inter_token_latency: BenchmarkMetric | None = None

output_token_throughput: BenchmarkMetric | None = None

request_latency: BenchmarkMetric | None = None

request_throughput: BenchmarkMetric | None = None

time_to_first_token: BenchmarkMetric | None = None

class sagemaker.serve.ai_inference_recommender.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>)

Bases: object

Parsed result of a completed benchmark job.

endpoint: str | None = None

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

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

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.ai_inference_recommender.FeatureGatedError(message: str = '', runbook_url: str = 'https://docs.aws.amazon.com/sagemaker/latest/dg/generative-ai-inference-recommendations.html')

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.ai_inference_recommender.InferenceFramework(value)

Bases: str, Enum

Inference framework to benchmark a recommendation against.

LMI = 'LMI'

VLLM = 'VLLM'

class sagemaker.serve.ai_inference_recommender.PerformanceTarget(value)

Bases: str, Enum

Optimization goal for a recommendation job.

COST = 'cost'

THROUGHPUT = 'throughput'

TTFT_MS = 'ttft-ms'

class sagemaker.serve.ai_inference_recommender.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())

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

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.ai_inference_recommender.Secret(arn: str, _created: bool = False, _session: Any | None = None)

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

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

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.ai_inference_recommender.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>)

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

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

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

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

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

Serialize the workload to a JSON string.

Secret values are flattened to their ARN strings.

tooling: Dict[str, Any]

exception sagemaker.serve.ai_inference_recommender.WorkloadValidationError(message='', **kwargs)

Bases: ValidationError

Raised when the server rejects a workload spec.

fmt = 'Server rejected workload: {message}'

sagemaker.serve.ai_inference_recommender.start_benchmark(endpoint: Endpoint | str, workload: Workload | str | None = None, *, output_path: str | None = None, role: str | None = None, inference_components: List[str] | None = None, vpc_config: VpcConfig | None = None, tags: List[Tag] | None = None, name: str | None = None, workload_config_name: str | None = None, sagemaker_session: Session | None = None, wait: bool = True, **workload_kwargs: Any) → AIBenchmarkJob

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.

Modules

exceptions	Exceptions for the AI inference recommender module.
jobs	Job subclasses that add show_result to the inference recommender resources.
result	Parsing of benchmark output artifacts from S3.
secrets	Helper for creating AWS Secrets Manager secrets.
workload	Workload spec builder.