diff --git a/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/__init__.py b/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/__init__.py index cad6f696d0..b3a3f166d6 100644 --- a/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/__init__.py +++ b/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/__init__.py @@ -25,11 +25,17 @@ BenchmarkJob, RecommendationJob, ) +from sagemaker.serve.ai_inference_recommender.listing import ( + list_benchmarks, + list_recommendations, +) from sagemaker.serve.ai_inference_recommender.result import ( + BenchmarkComparison, BenchmarkMetric, BenchmarkMetrics, BenchmarkResult, BenchmarkSearchResult, + compare_benchmarks, ) from sagemaker.serve.ai_inference_recommender.secrets import Secret from sagemaker.serve.ai_inference_recommender.workload import Workload @@ -39,6 +45,7 @@ __all__ = [ + "BenchmarkComparison", "BenchmarkJob", "BenchmarkMetric", "BenchmarkMetrics", @@ -51,5 +58,8 @@ "Secret", "Workload", "WorkloadValidationError", + "compare_benchmarks", + "list_benchmarks", + "list_recommendations", "start_benchmark", ] diff --git a/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/_recommendation_view.py b/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/_recommendation_view.py index af1ebb550f..e31eeb4ce3 100644 --- a/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/_recommendation_view.py +++ b/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/_recommendation_view.py @@ -24,6 +24,7 @@ _fmt_number, _format_table, _indent, + _require_pandas, ) @@ -63,9 +64,7 @@ def stats(self) -> Dict[str, float]: return dict(self._stats) def __repr__(self) -> str: - parts = ", ".join( - f"{stat}={_fmt_number(v)}" for stat, v in self._stats.items() - ) + parts = ", ".join(f"{stat}={_fmt_number(v)}" for stat, v in self._stats.items()) unit = f" {self.unit}" if self.unit else "" return f"<{parts}{unit}>" @@ -82,9 +81,7 @@ class _ExpectedPerformanceView: __slots__ = ("_by_metric",) def __init__(self, raw_rows: Optional[List[Any]]): - by_metric: Dict[str, Dict[str, Any]] = defaultdict( - lambda: {"unit": None, "stats": {}} - ) + by_metric: Dict[str, Dict[str, Any]] = defaultdict(lambda: {"unit": None, "stats": {}}) for row in raw_rows or []: metric = getattr(row, "metric", None) if not metric: @@ -139,9 +136,9 @@ def __len__(self) -> int: return len(self._by_metric) def __repr__(self) -> str: - return "{" + ", ".join( - f"{name}: {metric!r}" for name, metric in self._by_metric.items() - ) + "}" + return ( + "{" + ", ".join(f"{name}: {metric!r}" for name, metric in self._by_metric.items()) + "}" + ) def _to_float(value): @@ -219,12 +216,14 @@ def __str__(self) -> str: perf_rows = [] for m in ep: - perf_rows.append([ - _safe_str(m, "metric"), - _safe_str(m, "stat"), - _fmt_number(_safe_float(m, "value")), - _safe_str(m, "unit"), - ]) + perf_rows.append( + [ + _safe_str(m, "metric"), + _safe_str(m, "stat"), + _fmt_number(_safe_float(m, "value")), + _safe_str(m, "unit"), + ] + ) perf_table = _format_table( headers=["metric", "stat", "value", "unit"], rows=perf_rows, @@ -249,6 +248,31 @@ def _repr_pretty_(self, p, cycle): # Render the full table in notebooks (Jupyter uses this hook). p.text("..." if cycle else str(self)) + def _perf_records(self) -> List[Dict[str, Any]]: + """(metric, stat, value, unit) records for this row's expected + performance — the rows of the printed table, one per (metric, stat).""" + ep = getattr(self._raw, "expected_performance", None) or [] + return [ + { + "metric": getattr(m, "metric", None), + "stat": getattr(m, "stat", None), + "value": _safe_float(m, "value"), + "unit": getattr(m, "unit", None), + } + for m in ep + ] + + def to_dataframe(self): + """Return this recommendation's expected performance as a pandas + ``DataFrame`` — the same ``metric``/``stat``/``value``/``unit`` rows the + printed ``expected performance`` table shows, one row per (metric, stat). + + Requires pandas (an optional dependency); raises ``ImportError`` with + install guidance if it is missing. + """ + pd = _require_pandas() + return pd.DataFrame(self._perf_records(), columns=["metric", "stat", "value", "unit"]) + def _safe_str(obj, attr) -> str: if obj is None: @@ -291,41 +315,72 @@ def _repr_pretty_(self, p, cycle): # Render the full table in notebooks (Jupyter uses this hook). p.text("..." if cycle else str(self)) + # Column labels for the comparative table / DataFrame, in display order. + _TABLE_COLUMNS = ( + "idx", + "spec_name", + "instance_type", + "instances", + "copies/inst", + "container", + "req/s", + "tok/s", + "lat_p50", + "lat_p90", + "lat_p99", + "ttft_p50", + "itl_p50", + ) + + def _row_records(self) -> List[Dict[str, Any]]: + """One record per recommendation row, keyed by ``_TABLE_COLUMNS``. + + Values are kept native (numbers stay numbers, ``None`` stays ``None``) + so ``to_dataframe()`` gets real numeric columns; ``__str__`` formats + them for display. Shared so the printed table and the frame stay in + sync.""" + records = [] + for view in self: + dc = getattr(view.raw, "deployment_configuration", None) + ep = view.expected_performance + records.append( + { + "idx": view._index, + "spec_name": view.recommendation_spec_name, + "instance_type": getattr(dc, "instance_type", None) if dc else None, + "instances": getattr(dc, "instance_count", None) if dc else None, + "copies/inst": (getattr(dc, "copy_count_per_instance", None) if dc else None), + "container": _short_container_tag(_safe_str(dc, "image_uri")), + "req/s": _get_metric_stat(ep, "request_throughput", "avg"), + "tok/s": _get_metric_stat(ep, "output_token_throughput", "avg"), + "lat_p50": _get_metric_stat(ep, "request_latency", "p50"), + "lat_p90": _get_metric_stat(ep, "request_latency", "p90"), + "lat_p99": _get_metric_stat(ep, "request_latency", "p99"), + "ttft_p50": _get_metric_stat(ep, "time_to_first_token", "p50"), + "itl_p50": _get_metric_stat(ep, "inter_token_latency", "p50"), + } + ) + return records + def __str__(self) -> str: if not self: return "Recommendations[0] (no rows)" + _NUMERIC = {"req/s", "tok/s", "lat_p50", "lat_p90", "lat_p99", "ttft_p50", "itl_p50"} rows = [] - for view in self: - dc = getattr(view.raw, "deployment_configuration", None) - ep = view.expected_performance - rows.append([ - f"[{view._index}]", - view.recommendation_spec_name or "-", - _safe_str(dc, "instance_type"), - _safe_str(dc, "instance_count"), - _safe_str(dc, "copy_count_per_instance"), - _short_container_tag(_safe_str(dc, "image_uri")), - _fmt_number(_get_metric_stat(ep, "request_throughput", "avg")), - _fmt_number(_get_metric_stat(ep, "output_token_throughput", "avg")), - _fmt_number(_get_metric_stat(ep, "request_latency", "p50")), - _fmt_number(_get_metric_stat(ep, "request_latency", "p90")), - _fmt_number(_get_metric_stat(ep, "request_latency", "p99")), - _fmt_number(_get_metric_stat(ep, "time_to_first_token", "p50")), - _fmt_number(_get_metric_stat(ep, "inter_token_latency", "p50")), - ]) - - table = _format_table( - headers=[ - "idx", "spec_name", "instance_type", - "instances", "copies/inst", - "container", - "req/s", "tok/s", - "lat_p50", "lat_p90", "lat_p99", - "ttft_p50", "itl_p50", - ], - rows=rows, - ) + for rec in self._row_records(): + row = [] + for col in self._TABLE_COLUMNS: + value = rec[col] + if col == "idx": + row.append(f"[{value}]") + elif col in _NUMERIC: + row.append(_fmt_number(value)) + else: + row.append("-" if value in (None, "") else str(value)) + rows.append(row) + + table = _format_table(headers=list(self._TABLE_COLUMNS), rows=rows) return ( f"Recommendations[{len(self)}] (.best = top row; index by [N] for full detail)\n" @@ -333,6 +388,29 @@ def __str__(self) -> str: f"lat/ttft/itl in ms; req/s = requests/sec; tok/s = output tokens/sec" ) + def to_dataframe(self): + """Return the recommendations as a pandas ``DataFrame`` — one row per + recommendation, columns matching the printed comparative table + (``instance_type``, ``instances``, ``req/s``, ``lat_p50``, ...), indexed + by the recommendation index ``idx``. + + Latency columns are milliseconds; ``req/s`` is requests/sec and ``tok/s`` + is output tokens/sec, same as the printed table's footnote. Numeric + columns hold real numbers (``NaN`` where a metric is absent), not + preformatted strings. + + Requires pandas (an optional dependency); raises ``ImportError`` with + install guidance if it is missing. + """ + pd = _require_pandas() + columns = [c for c in self._TABLE_COLUMNS if c != "idx"] + records = self._row_records() + return pd.DataFrame( + [{c: rec[c] for c in columns} for rec in records], + columns=columns, + index=pd.Index([rec["idx"] for rec in records], name="idx"), + ) + def _get_metric_stat(ep_view, metric_name: str, stat: str): try: diff --git a/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/listing.py b/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/listing.py new file mode 100644 index 0000000000..be9e04bc92 --- /dev/null +++ b/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/listing.py @@ -0,0 +1,207 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"). You +# may not use this file except in compliance with the License. A copy of +# the License is located at +# +# http://aws.amazon.com/apache2.0/ +# +# or in the "license" file accompanying this file. This file is +# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF +# ANY KIND, either express or implied. See the License for the specific +# language governing permissions and limitations under the License. +"""List and filter AI benchmark / recommendation jobs. + +The underlying ``ListAIBenchmarkJobs`` / ``ListAIRecommendationJobs`` APIs only +filter by name substring, status, and creation-time window — the endpoint a +benchmark targeted, or the model a recommendation ran on, live on the full +Describe response, not the list summary. So ``endpoint`` / ``model`` / +``model_package`` filtering is done client-side: the native filters narrow the +list server-side, then each candidate is described (hydrated) and matched on the +nested field. ``max_results`` bounds that Describe fan-out. +""" +from __future__ import absolute_import + +import logging +from typing import List, Optional + +from sagemaker.core.helper.session_helper import Session +from sagemaker.core.telemetry.constants import Feature +from sagemaker.core.telemetry.telemetry_logging import _telemetry_emitter +from sagemaker.serve.ai_inference_recommender.jobs import BenchmarkJob, RecommendationJob + +logger = logging.getLogger(__name__) + +# Default cap on how many jobs are described when a client-side filter +# (endpoint / model / model_package) is applied. Keeps a broad list from +# fanning out into an unbounded number of Describe calls. +DEFAULT_MAX_RESULTS = 100 + + +def _endpoint_matches(job, endpoint: str) -> bool: + """True if a benchmark job targeted ``endpoint`` (by name or ARN).""" + target = getattr(job, "benchmark_target", None) + ep = getattr(target, "endpoint", None) if target else None + identifier = getattr(ep, "identifier", None) if ep else None + if not identifier: + return False + # identifier may be a name or an ARN; match either the exact value or the + # name suffix of an ARN (endpoint/). + return endpoint == identifier or identifier.endswith(f"/{endpoint}") + + +def _model_matches(job, model: str) -> bool: + """True if a recommendation job ran on ``model`` (its source S3 URI).""" + source = getattr(job, "model_source", None) + s3 = getattr(source, "s3", None) if source else None + s3_uri = getattr(s3, "s3_uri", None) if s3 else None + if not s3_uri: + return False + return model == s3_uri or s3_uri.rstrip("/") == model.rstrip("/") + + +def _model_package_matches(job, model_package: str) -> bool: + """True if a recommendation job is associated with ``model_package``. + + Matches either the output model-package group the job registers into, or a + model-package ARN produced on one of the job's recommendation rows. + """ + output = getattr(job, "output_config", None) + group = getattr(output, "model_package_group_identifier", None) if output else None + if group and (model_package == group or group.endswith(f"/{model_package}")): + return True + for row in getattr(job, "recommendations", None) or []: + details = getattr(row, "model_details", None) + arn = getattr(details, "model_package_arn", None) if details else None + if arn and (model_package == arn or arn.endswith(f"/{model_package}")): + return True + return False + + +def _collect(iterator, predicate, max_results: int, subclass) -> list: + """Describe candidates from ``iterator`` and keep those matching ``predicate``. + + Each candidate is re-typed to ``subclass`` (so ``show_result`` is available) + and, when a predicate is given, refreshed to hydrate the nested fields the + list summary omits. Stops once ``max_results`` matches are collected; logs + when the scan is truncated so a silent cap never reads as "all matches". + """ + matches: list = [] + scanned = 0 + for job in iterator: + job.__class__ = subclass + if predicate is None: + matches.append(job) + else: + # The list summary lacks the nested field; hydrate before matching. + try: + job.refresh() + except Exception as exc: # pragma: no cover - best-effort hydration + logger.debug("Skipping %s; could not describe it: %s", job, exc) + continue + if predicate(job): + matches.append(job) + if len(matches) >= max_results: + logger.info( + "Reached max_results=%d; stopping the scan. Raise max_results to " + "list more.", + max_results, + ) + break + scanned += 1 + return matches + + +@_telemetry_emitter( + feature=Feature.INFERENCE_RECOMMENDER, func_name="ai_inference_recommender.list_benchmarks" +) +def list_benchmarks( + *, + endpoint: Optional[str] = None, + status: Optional[str] = None, + name_contains: Optional[str] = None, + max_results: int = DEFAULT_MAX_RESULTS, + sagemaker_session: Optional[Session] = None, +) -> List[BenchmarkJob]: + """List benchmark jobs, optionally filtered by the endpoint they targeted. + + Args: + endpoint: Endpoint name or ARN. Client-side filter (each candidate is + described, since the endpoint is not on the list summary). + status: ``StatusEquals`` filter, applied server-side. + name_contains: ``NameContains`` filter, applied server-side. + max_results: Cap on the number of jobs returned (and, when ``endpoint`` + is set, on how many are described). Defaults to + ``DEFAULT_MAX_RESULTS``. + sagemaker_session: Optional session; a default is created if omitted. + + Returns: + A list of ``BenchmarkJob`` (newest first), each with ``show_result``. + """ + iterator = BenchmarkJob.get_all( + **_native_filters(name_contains, status), + sort_by="CreationTime", + sort_order="Descending", + session=sagemaker_session, + ) + predicate = (lambda job: _endpoint_matches(job, endpoint)) if endpoint else None + return _collect(iterator, predicate, max_results, BenchmarkJob) + + +@_telemetry_emitter( + feature=Feature.INFERENCE_RECOMMENDER, + func_name="ai_inference_recommender.list_recommendations", +) +def list_recommendations( + *, + model: Optional[str] = None, + model_package: Optional[str] = None, + status: Optional[str] = None, + name_contains: Optional[str] = None, + max_results: int = DEFAULT_MAX_RESULTS, + sagemaker_session: Optional[Session] = None, +) -> List[RecommendationJob]: + """List recommendation jobs, optionally filtered by model or model package. + + Args: + model: Model source S3 URI the job ran on. Client-side filter. + model_package: Model-package ARN or group identifier associated with the + job (its output group, or a package produced on a recommendation + row). Client-side filter. + status: ``StatusEquals`` filter, applied server-side. + name_contains: ``NameContains`` filter, applied server-side. + max_results: Cap on the number of jobs returned (and, when a client-side + filter is set, on how many are described). Defaults to + ``DEFAULT_MAX_RESULTS``. + sagemaker_session: Optional session; a default is created if omitted. + + Returns: + A list of ``RecommendationJob`` (newest first), each with ``show_result``. + """ + if model and model_package: + raise ValueError( + "Pass only one of `model` or `model_package` to list_recommendations()." + ) + iterator = RecommendationJob.get_all( + **_native_filters(name_contains, status), + sort_by="CreationTime", + sort_order="Descending", + session=sagemaker_session, + ) + predicate = None + if model: + predicate = lambda job: _model_matches(job, model) # noqa: E731 + elif model_package: + predicate = lambda job: _model_package_matches(job, model_package) # noqa: E731 + return _collect(iterator, predicate, max_results, RecommendationJob) + + +def _native_filters(name_contains: Optional[str], status: Optional[str]) -> dict: + """Build the server-side ``get_all`` filter kwargs, omitting unset ones so + each defaults to the sagemaker-core ``Unassigned`` sentinel.""" + kwargs = {} + if name_contains: + kwargs["name_contains"] = name_contains + if status: + kwargs["status_equals"] = status + return kwargs diff --git a/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/result.py b/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/result.py index 132313e691..dddc593622 100644 --- a/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/result.py +++ b/sagemaker-serve/src/sagemaker/serve/ai_inference_recommender/result.py @@ -17,7 +17,7 @@ import json import tarfile from dataclasses import dataclass, field -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional, Sequence from urllib.parse import urlparse import boto3 @@ -29,6 +29,11 @@ # archive is how we tell a sweep run apart from a single run. SEARCH_HISTORY_FILENAME = "search_history.json" +# Statistic columns exposed by ``to_dataframe()``. Superset of the printed +# table's avg/p50/p90/p99 — the DataFrame also carries min/max/p95/stddev, +# which the text table omits for width. +_METRIC_STAT_COLUMNS = ("avg", "min", "max", "p50", "p90", "p95", "p99", "stddev") + @dataclass class BenchmarkMetric: @@ -96,6 +101,23 @@ def _repr_pretty_(self, p, cycle): # Render the full table in notebooks (Jupyter uses this hook). p.text("..." if cycle else str(self)) + def to_dataframe(self): + """Return the metrics as a pandas ``DataFrame`` indexed by metric name. + + One row per metric, one column per statistic (``unit`` plus + ``avg``/``min``/``max``/``p50``/``p90``/``p95``/``p99``/``stddev``). + Rows are ordered exactly as the printed table: non-HTTP metrics + alphabetically, then ``http_*`` transport metrics last. Unlike the text + table, the frame keeps ``min``/``max``/``p95``/``stddev``. + + Requires pandas (an optional dependency); raises ``ImportError`` with + install guidance if it is missing. + """ + pd = _require_pandas() + ordered_names = [n for n in sorted(self.all_metrics) if not n.startswith("http_")] + ordered_names += [n for n in sorted(self.all_metrics) if n.startswith("http_")] + return _metrics_dataframe(pd, [(n, self.all_metrics[n]) for n in ordered_names]) + @classmethod def from_profile_json(cls, profile: Dict[str, Any]) -> "BenchmarkMetrics": all_metrics: Dict[str, BenchmarkMetric] = {} @@ -239,9 +261,22 @@ def __str__(self) -> str: f" s3_output_location: {self.s3_output_location}\n" f" search:\n{_indent(str(self.search), ' ')}" ) - # Order: well-known headline metrics first, then everything else - # alphabetized, then HTTP-level transport metrics last (they're - # noise for most readers, useful only for debugging). + table = _format_metrics_table(self._ordered_metric_pairs()) + return ( + f"BenchmarkResult\n" + f" endpoint: {self.endpoint or '-'}\n" + f" workload_config: {self.workload_config or '-'}\n" + f" tool_version: {self.tool_version or '-'}\n" + f" s3_output_location: {self.s3_output_location}\n" + f" metrics:\n{_indent(table, ' ')}\n" + f" raw profile available via .profile" + ) + + def _ordered_metric_pairs(self): + """(name, metric) pairs in display order: well-known headline metrics + first (canonical order), then the rest alphabetized, then ``http_*`` + transport metrics last. Shared by ``__str__`` and ``to_dataframe()`` so + the printed table and the frame stay in the same order.""" seen = set() headline = [] for name in _KEY_METRIC_FIELDS: @@ -256,18 +291,29 @@ def __str__(self) -> str: continue bucket = http if name.startswith("http_") else rest bucket.append((name, self.metrics.all_metrics[name])) + return headline + rest + http - ordered = headline + rest + http - table = _format_metrics_table(ordered) - return ( - f"BenchmarkResult\n" - f" endpoint: {self.endpoint or '-'}\n" - f" workload_config: {self.workload_config or '-'}\n" - f" tool_version: {self.tool_version or '-'}\n" - f" s3_output_location: {self.s3_output_location}\n" - f" metrics:\n{_indent(table, ' ')}\n" - f" raw profile available via .profile" - ) + def to_dataframe(self): + """Return this result's metrics as a pandas ``DataFrame``. + + One row per metric (indexed by metric name), one column per statistic — + the same shape as :meth:`BenchmarkMetrics.to_dataframe`, but ordered as + this result prints: headline metrics first, then the rest alphabetized, + then ``http_*`` transport metrics last. + + A search/sweep result has no single metric profile; call + ``result.search`` for its outcome instead. + + Requires pandas (an optional dependency); raises ``ImportError`` with + install guidance if it is missing. + """ + pd = _require_pandas() + if self.search is not None: + raise ValueError( + "This is a search/sweep result with no single metric profile to " + "tabulate. Inspect result.search for the sweep outcome instead." + ) + return _metrics_dataframe(pd, self._ordered_metric_pairs()) def __repr__(self) -> str: return ( @@ -319,9 +365,7 @@ def from_job( f"(status={status}). Call job.wait() (or pass wait=True to " f"start_benchmark) before BenchmarkResult.from_job()." ) - if job.output_config is None or not getattr( - job.output_config, "s3_output_location", None - ): + if job.output_config is None or not getattr(job.output_config, "s3_output_location", None): failure_reason = getattr(job, "failure_reason", None) hint = ( f"Job failed: {failure_reason or 'no reason provided'}." @@ -449,9 +493,7 @@ def _find_object(s3_client, bucket: str, prefix: str, suffix: str) -> str: key = obj.get("Key", "") if key.endswith(suffix): return key - raise FileNotFoundError( - f"No object ending in {suffix!r} under s3://{bucket}/{prefix}" - ) + raise FileNotFoundError(f"No object ending in {suffix!r} under s3://{bucket}/{prefix}") def _read_member_from_tar_gz(archive_bytes: bytes, suffix: str) -> Optional[bytes]: @@ -473,6 +515,24 @@ def _as_float(value: Any) -> Optional[float]: return None +def _require_pandas(): + """Import pandas lazily, only when ``to_dataframe()`` is actually called. + + pandas reaches this module transitively (via ``sagemaker-core``), but this + module keeps itself dependency-light on purpose — the printed tables are + stdlib-only. So import pandas on demand rather than at module load, and + raise a clear, actionable error on the off chance it is unavailable. + """ + try: + import pandas as pd # noqa: F401 (local import by design) + except ImportError as exc: # pragma: no cover - trivial re-raise + raise ImportError( + "to_dataframe() requires pandas, which is not installed. " + "Install it with `pip install pandas`." + ) from exc + return pd + + def _fmt_number(value: Optional[float]) -> str: """Render a number compact for the metrics table; '-' for None.""" if value is None: @@ -486,18 +546,41 @@ def _indent(text: str, prefix: str) -> str: return "\n".join(prefix + line if line else line for line in text.splitlines()) +def _metrics_dataframe(pd, name_metric_pairs): + """Build a metric-indexed DataFrame from (name, BenchmarkMetric) pairs. + + Columns are ``unit`` plus every stat in ``_METRIC_STAT_COLUMNS``. Row order + follows ``name_metric_pairs`` (the caller decides ordering); an empty input + yields an empty frame with the right columns so callers can rely on the + schema. + """ + columns = ["unit", *_METRIC_STAT_COLUMNS] + names = [name for name, _ in name_metric_pairs] + data = [ + { + "unit": metric.unit, + **{stat: getattr(metric, stat, None) for stat in _METRIC_STAT_COLUMNS}, + } + for _name, metric in name_metric_pairs + ] + frame = pd.DataFrame(data, columns=columns, index=pd.Index(names, name="metric")) + return frame + + def _format_metrics_table(name_metric_pairs) -> str: """Render an iterable of (name, BenchmarkMetric) pairs as a table.""" rows = [] for _name, metric in name_metric_pairs: - rows.append([ - metric.name, - metric.unit or "-", - _fmt_number(metric.avg), - _fmt_number(metric.p50), - _fmt_number(metric.p90), - _fmt_number(metric.p99), - ]) + rows.append( + [ + metric.name, + metric.unit or "-", + _fmt_number(metric.avg), + _fmt_number(metric.p50), + _fmt_number(metric.p90), + _fmt_number(metric.p99), + ] + ) return _format_table( headers=["metric", "unit", "avg", "p50", "p90", "p99"], rows=rows, @@ -524,7 +607,195 @@ def _format_table(headers, rows) -> str: header_line = " ".join(str(h).ljust(widths[i]) for i, h in enumerate(headers)) sep_line = " ".join("─" * widths[i] for i in range(len(headers))) body = "\n".join( - " ".join(cell.ljust(widths[i]) for i, cell in enumerate(row)) - for row in str_rows + " ".join(cell.ljust(widths[i]) for i, cell in enumerate(row)) for row in str_rows ) return f"{header_line}\n{sep_line}\n{body}" + + +# Metrics where a higher value is better (throughput); everything else in +# _KEY_METRIC_FIELDS is a latency/duration where lower is better. Used to label +# each run's delta vs. the baseline as an improvement or a regression. +_HIGHER_IS_BETTER = frozenset( + { + "request_throughput", + "output_token_throughput", + "e2e_output_token_throughput", + } +) + + +@dataclass +class BenchmarkComparison: + """Side-by-side comparison of two or more ``BenchmarkResult`` runs. + + The first run is the baseline; each other run's ``__str__`` shows its value + for every key metric alongside the percentage change from the baseline, + signed so a ``+`` is always an improvement (higher throughput / lower + latency) regardless of the metric's direction. + + Attributes: + results: the compared results, baseline first. + names: display label per result (defaults to ``run1``, ``run2``, ...). + stat: which per-metric statistic is compared (``avg`` by default; any of + ``avg``/``p50``/``p90``/``p95``/``p99``/``min``/``max``). + """ + + results: List["BenchmarkResult"] + names: List[str] + stat: str = "avg" + + def _metric_names(self) -> List[str]: + """Key metrics first (in canonical order), then any other metric present + in at least one run — so the table covers everything, headline first.""" + ordered = [ + name + for name in _KEY_METRIC_FIELDS + if any(name in r.metrics.all_metrics for r in self.results) + ] + seen = set(ordered) + for r in self.results: + for name in sorted(r.metrics.all_metrics): + if name not in seen: + ordered.append(name) + seen.add(name) + return ordered + + def _value(self, result: "BenchmarkResult", metric_name: str) -> Optional[float]: + metric = result.metrics.all_metrics.get(metric_name) + return getattr(metric, self.stat, None) if metric is not None else None + + def _delta_value(self, metric_name: str, baseline, value) -> Optional[float]: + """Signed percentage change vs. baseline, oriented so + is better. + + Returns ``None`` when a delta is undefined (missing value or zero + baseline). ``__str__`` formats it; ``to_dataframe()`` keeps it numeric. + """ + if baseline is None or value is None or baseline == 0: + return None + pct = (value - baseline) / abs(baseline) * 100.0 + if metric_name not in _HIGHER_IS_BETTER: + # Lower-is-better metric: flip the sign so a drop reads as +. + pct = -pct + return pct + + def _delta_cell(self, metric_name: str, baseline, value) -> str: + """Signed percentage change vs. baseline as a display string.""" + pct = self._delta_value(metric_name, baseline, value) + return "-" if pct is None else f"{pct:+.1f}%" + + def _unit_for(self, metric_name: str) -> Optional[str]: + for r in self.results: + m = r.metrics.all_metrics.get(metric_name) + if m is not None and m.unit: + return m.unit + return None + + def __str__(self) -> str: + metric_names = self._metric_names() + if not metric_names: + return "BenchmarkComparison (no metrics to compare)" + + # Columns: metric, unit, one value column per run, and a Δ% column per + # non-baseline run (vs. the baseline, the first run). + headers = ["metric", "unit"] + headers += list(self.names) + headers += [f"Δ% {name}" for name in self.names[1:]] + + rows = [] + for metric_name in metric_names: + unit = self._unit_for(metric_name) or "-" + values = [self._value(r, metric_name) for r in self.results] + row = [metric_name, unit] + [_fmt_number(v) for v in values] + baseline = values[0] + row += [self._delta_cell(metric_name, baseline, v) for v in values[1:]] + rows.append(row) + + table = _format_table(headers=headers, rows=rows) + baseline_note = f"baseline: {self.names[0]} | stat: {self.stat} (+Δ = better)" + return f"BenchmarkComparison\n {baseline_note}\n{_indent(table, ' ')}" + + def to_dataframe(self): + """Return the comparison as a pandas ``DataFrame``. + + Mirrors the printed table: one row per metric (indexed by metric name), + a ``unit`` column, one column per run (named by :attr:`names`, holding + the compared ``stat``), and a ``Δ% `` column per non-baseline run — + signed so ``+`` is always an improvement. Delta values are numeric + percentages (``NaN`` where undefined), not preformatted strings. + + Requires pandas (an optional dependency); raises ``ImportError`` with + install guidance if it is missing. + """ + pd = _require_pandas() + columns = ["unit", *self.names, *(f"Δ% {name}" for name in self.names[1:])] + metric_names = self._metric_names() + data = [] + for metric_name in metric_names: + values = [self._value(r, metric_name) for r in self.results] + baseline = values[0] + record = {"unit": self._unit_for(metric_name)} + for name, value in zip(self.names, values): + record[name] = value + for name, value in zip(self.names[1:], values[1:]): + record[f"Δ% {name}"] = self._delta_value(metric_name, baseline, value) + data.append(record) + return pd.DataFrame(data, columns=columns, index=pd.Index(metric_names, name="metric")) + + def __repr__(self) -> str: + return ( + f"BenchmarkComparison({len(self.results)} runs: " + f"{', '.join(self.names)}; print() for the table)" + ) + + def _repr_pretty_(self, p, cycle): + p.text("..." if cycle else str(self)) + + +def compare_benchmarks( + *results: "BenchmarkResult", + names: Optional[Sequence[str]] = None, + stat: str = "avg", +) -> BenchmarkComparison: + """Compare two or more benchmark runs and return a tabular comparison. + + The first result is the baseline; each subsequent run is reported with a + signed percentage change from it (oriented so ``+`` is always better — + higher throughput or lower latency). ``print()`` the returned object for the + table. + + Args: + *results: two or more ``BenchmarkResult`` objects (from + ``job.show_result()``). The first is the baseline. + names: optional display label per result; defaults to ``run1``, + ``run2``, .... Must match the number of results when given. + stat: which per-metric statistic to compare — one of ``avg`` (default), + ``p50``, ``p90``, ``p95``, ``p99``, ``min``, ``max``. + + Returns: + BenchmarkComparison: renders a metric-by-run table with per-run deltas. + + Raises: + ValueError: if fewer than two results are given, if ``names`` length + does not match, if ``stat`` is not a known statistic, or if any + result is a concurrency-search/sweep run (which has no single metric + profile to compare). + """ + if len(results) < 2: + raise ValueError("compare_benchmarks() needs at least two results to compare.") + if any(r.is_search for r in results): + raise ValueError( + "compare_benchmarks() compares single-run results; a search/sweep " + "result has no single metric profile. Compare the winning runs instead." + ) + valid_stats = {"avg", "p50", "p90", "p95", "p99", "min", "max"} + if stat not in valid_stats: + raise ValueError(f"stat must be one of {sorted(valid_stats)}, got {stat!r}.") + if names is not None: + if len(names) != len(results): + raise ValueError( + f"names has {len(names)} entries but {len(results)} results were given." + ) + labels = list(names) + else: + labels = [f"run{i + 1}" for i in range(len(results))] + return BenchmarkComparison(results=list(results), names=labels, stat=stat) diff --git a/sagemaker-serve/src/sagemaker/serve/model_builder.py b/sagemaker-serve/src/sagemaker/serve/model_builder.py index 11853c8a11..7b5b626770 100644 --- a/sagemaker-serve/src/sagemaker/serve/model_builder.py +++ b/sagemaker-serve/src/sagemaker/serve/model_builder.py @@ -5424,6 +5424,7 @@ def deploy( # generate_deployment_recommendations was called previously, or when # this builder was hydrated via ModelBuilder.from_recommendation_job(...). use_recommendation: Optional[bool] = None, + recommendation: Optional[Any] = None, recommendation_index: int = 0, recommendation_spec_name: Optional[str] = None, auto_approve: bool = False, @@ -5467,6 +5468,10 @@ def deploy( 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 (optional): Recommendation deploy only. A recommendation row to + deploy, e.g. ``mb.recommendations.best`` or ``mb.recommendations[i]``. Use + this instead of ``recommendation_index`` / ``recommendation_spec_name`` to + deploy a row without hand-copying its index. Mutually exclusive with those two. 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. @@ -5519,6 +5524,22 @@ def deploy( "Call generate_deployment_recommendations(...) or build via " "ModelBuilder.from_recommendation_job(...) first." ) + # A recommendation row object (e.g. mb.recommendations.best or + # mb.recommendations[i]) can be passed directly instead of an index or + # spec name. Resolve it to the spec name the selection logic already + # understands (falling back to its positional index when the row has no + # spec name), so callers never hand-copy a magic index. + if recommendation is not None: + if recommendation_spec_name is not None or recommendation_index: + raise ValueError( + "Pass only one of `recommendation`, `recommendation_spec_name`, " + "or `recommendation_index` to deploy()." + ) + resolved_spec = getattr(recommendation, "recommendation_spec_name", None) + if resolved_spec is not None: + recommendation_spec_name = resolved_spec + else: + recommendation_index = getattr(recommendation, "_index", 0) if has_recommendation and use_recommendation is not False: return self._deploy_recommendation( recommendation_index=recommendation_index, diff --git a/sagemaker-serve/tests/integ/test_ai_inference_recommender_enhancements_integration.py b/sagemaker-serve/tests/integ/test_ai_inference_recommender_enhancements_integration.py new file mode 100644 index 0000000000..6effe795d1 --- /dev/null +++ b/sagemaker-serve/tests/integ/test_ai_inference_recommender_enhancements_integration.py @@ -0,0 +1,296 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"). You +# may not use this file except in compliance with the License. A copy of +# the License is located at +# +# http://aws.amazon.com/apache2.0/ +# +# or in the "license" file accompanying this file. This file is +# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF +# ANY KIND, either express or implied. See the License for the specific +# language governing permissions and limitations under the License. +"""End-to-end integration tests for the inference-recommender enhancements: +``list_benchmarks`` / ``list_recommendations`` filtering, ``deploy`` from a +recommendation row (``mb.recommendations.best``), and ``compare_benchmarks``. +""" +from __future__ import absolute_import + +import logging +import time +import uuid + +import pytest + +from sagemaker.core.helper.session_helper import Session, get_execution_role +from sagemaker.core.jumpstart.configs import JumpStartConfig +from sagemaker.core.resources import ( + AIBenchmarkJob, + AIRecommendationJob, + AIWorkloadConfig, + EndpointConfig, + Model, + ModelPackage, +) +from sagemaker.serve.ai_inference_recommender import ( + Workload, + compare_benchmarks, + list_benchmarks, + list_recommendations, + start_benchmark, +) +from sagemaker.serve.model_builder import ModelBuilder +from sagemaker.train.configs import Compute + +logger = logging.getLogger(__name__) + +MODEL_ID = "huggingface-reasoning-qwen3-06b" +INSTANCE_TYPE = "ml.g6.2xlarge" +WORKLOAD_TOKENIZER = "gpt2" + + +def _synthetic_workload(): + return Workload.synthetic( + tokenizer=WORKLOAD_TOKENIZER, + concurrency=1, + request_count=10, + prompt_input_tokens_mean=32, + output_tokens_mean=32, + streaming=True, + ) + + +def test_list_benchmarks_and_recommendations_plumbing(): + """The listing helpers execute against the live API (list + describe + + client-side filter) and a filter with no match returns an empty list. + + No GPU: this exercises the ``get_all`` / describe / filter plumbing without + creating any resource, so it runs fast and independently of the e2e below. + """ + logger.info("Listing plumbing: list_benchmarks / list_recommendations ...") + + benches = list_benchmarks(max_results=5) + assert isinstance(benches, list) + logger.info("list_benchmarks() returned %d job(s)", len(benches)) + + recs = list_recommendations(max_results=5) + assert isinstance(recs, list) + logger.info("list_recommendations() returned %d job(s)", len(recs)) + + # A filter that cannot match returns an empty list (not an error). + no_match_ep = f"no-such-endpoint-{uuid.uuid4().hex}" + assert list_benchmarks(endpoint=no_match_ep, max_results=5) == [] + + no_match_model = f"s3://no-such-bucket-{uuid.uuid4().hex}/model/" + assert list_recommendations(model=no_match_model, max_results=5) == [] + logger.info("Non-matching filters correctly returned empty lists.") + + +@pytest.mark.slow_test +@pytest.mark.gpu_intensive +def test_recommendation_deploy_best_and_compare_e2e(): + """Full flow across all three enhancements, sharing one rec job + endpoint: + + 1. run a recommendation job, + 2. ``list_recommendations(model=...)`` finds it (client-side filter), + 3. ``deploy(recommendation=mb.recommendations.best)`` reaches InService, + 4. run two benchmarks against that endpoint, + 5. ``list_benchmarks(endpoint=...)`` finds them, + 6. ``compare_benchmarks`` renders a two-run comparison. + """ + unique_id = f"{int(time.time())}-{uuid.uuid4().hex[:8]}" + role = get_execution_role(sagemaker_session=Session()) + rec_job_name = f"air-enh-rec-{unique_id}" + rec_wl_name = f"air-enh-rec-wl-{unique_id}" + src_model_name = f"air-enh-src-{unique_id}" + dep_model_name = f"air-enh-model-{unique_id}" + dep_config_name = f"air-enh-cfg-{unique_id}" + endpoint_name = f"air-enh-ep-{unique_id}" + bench_a_name = f"air-enh-bench-a-{unique_id}" + bench_b_name = f"air-enh-bench-b-{unique_id}" + bench_a_wl = f"air-enh-bench-a-wl-{unique_id}" + bench_b_wl = f"air-enh-bench-b-wl-{unique_id}" + + source_model = None + endpoint = None + rec_model_package_arn = None + model_uri = None + + try: + mb = ModelBuilder.from_jumpstart_config( + jumpstart_config=JumpStartConfig(model_id=MODEL_ID), + compute=Compute(instance_type=INSTANCE_TYPE), + role_arn=role, + ) + source_model = mb.build(model_name=src_model_name) + model_uri = _model_source_uri(src_model_name) + + # (1) recommendation job + rec_job = mb.generate_deployment_recommendations( + workload=_synthetic_workload(), + performance_target="throughput", + instance_types=[INSTANCE_TYPE], + advanced_optimization=False, + framework="LMI", + role_arn=role, + job_name=rec_job_name, + workload_config_name=rec_wl_name, + wait=True, + ) + assert rec_job.ai_recommendation_job_status == "Completed", ( + f"Recommendation job did not complete: " + f"{rec_job.ai_recommendation_job_status} / " + f"{getattr(rec_job, 'failure_reason', None)}" + ) + rows = mb.recommendations + assert rows, "mb.recommendations is empty after a completed job" + rec_model_package_arn = getattr( + getattr(rows.best, "model_details", None), "model_package_arn", None + ) + logger.info("Recommendation complete; best spec=%s", rows.best.recommendation_spec_name) + + # (2) list_recommendations model filter returns rows that all match the + # requested model. We do NOT assert this specific just-created job is in + # the result: the model URI is a shared JumpStart cache path many jobs + # reuse, and freshly-created jobs are subject to List eventual + # consistency — so pinning on "find my exact job" here is racy. Instead + # assert the filter's contract: every returned job's model_source matches + # the requested URI. (Deterministic filter coverage lives in the unit + # tests and the no-GPU plumbing test.) + if model_uri: + found = list_recommendations(model=model_uri, max_results=25) + for job in found: + src = getattr(getattr(job, "model_source", None), "s3", None) + uri = getattr(src, "s3_uri", None) if src else None + assert uri is not None and uri.rstrip("/") == model_uri.rstrip("/"), ( + f"list_recommendations(model=...) returned a non-matching job: " + f"{job.ai_recommendation_job_name} has model_source {uri}" + ) + logger.info( + "list_recommendations(model=...) returned %d matching job(s).", len(found) + ) + + # (3) deploy the best recommendation row directly (no magic index) + endpoint = mb.deploy( + endpoint_name=endpoint_name, + recommendation=rows.best, + model_name=dep_model_name, + endpoint_config_name=dep_config_name, + role=role, + auto_approve=True, + wait=True, + ) + assert endpoint.endpoint_status == "InService", ( + f"Endpoint did not reach InService: {endpoint.endpoint_status}" + ) + logger.info("Deployed mb.recommendations.best -> %s InService", endpoint_name) + + # (4) two benchmarks against the endpoint + bench_a = start_benchmark( + endpoint=endpoint_name, + workload=_synthetic_workload(), + role=role, + name=bench_a_name, + workload_config_name=bench_a_wl, + wait=True, + ) + bench_b = start_benchmark( + endpoint=endpoint_name, + workload=_synthetic_workload(), + role=role, + name=bench_b_name, + workload_config_name=bench_b_wl, + wait=True, + ) + for job in (bench_a, bench_b): + assert job.ai_benchmark_job_status == "Completed", ( + f"Benchmark {job.get_name()} did not complete: " + f"{job.ai_benchmark_job_status}" + ) + + # (5) list_benchmarks(endpoint=...) filters by this run's (unique) + # endpoint. Hard-assert the filter contract: every returned job targets + # this endpoint. Finding both specific jobs is best-effort (List is + # eventually consistent for freshly-created jobs), logged not gated. + listed = list_benchmarks(endpoint=endpoint_name, max_results=25) + for job in listed: + target = getattr(job, "benchmark_target", None) + ep = getattr(target, "endpoint", None) if target else None + identifier = getattr(ep, "identifier", None) if ep else None + assert identifier and ( + identifier == endpoint_name or identifier.endswith(f"/{endpoint_name}") + ), ( + f"list_benchmarks(endpoint=...) returned a non-matching job: " + f"{job.ai_benchmark_job_name} targets {identifier}" + ) + listed_names = [j.ai_benchmark_job_name for j in listed] + both_present = bench_a_name in listed_names and bench_b_name in listed_names + logger.info( + "list_benchmarks(endpoint=...) returned %d job(s); both this run's " + "benchmarks present: %s", + len(listed), + both_present, + ) + + # (6) compare the two runs + result_a = bench_a.show_result() + result_b = bench_b.show_result() + comparison = compare_benchmarks(result_a, result_b, names=["run_a", "run_b"]) + rendered = str(comparison) + assert "BenchmarkComparison" in rendered + assert "run_a" in rendered and "run_b" in rendered + logger.info("compare_benchmarks rendered:\n%s", rendered) + + finally: + _delete_quietly( + lambda: Model.get(model_name=dep_model_name), f"Model {dep_model_name}" + ) + if endpoint is not None: + _delete_quietly(lambda: endpoint, f"Endpoint {endpoint_name}") + _delete_quietly( + lambda: EndpointConfig.get(endpoint_config_name=dep_config_name), + f"EndpointConfig {dep_config_name}", + ) + if source_model is not None: + _delete_quietly(lambda: source_model, f"Model {src_model_name}") + for job_name in (bench_a_name, bench_b_name): + _delete_quietly( + lambda n=job_name: AIBenchmarkJob.get(ai_benchmark_job_name=n), + f"AIBenchmarkJob {job_name}", + ) + _delete_quietly( + lambda: AIRecommendationJob.get(ai_recommendation_job_name=rec_job_name), + f"AIRecommendationJob {rec_job_name}", + ) + for wl in (rec_wl_name, bench_a_wl, bench_b_wl): + _delete_quietly( + lambda n=wl: AIWorkloadConfig.get(ai_workload_config_name=n), + f"AIWorkloadConfig {wl}", + ) + if rec_model_package_arn: + _delete_quietly( + lambda: ModelPackage.get(model_package_name=rec_model_package_arn), + f"ModelPackage {rec_model_package_arn}", + ) + + +def _model_source_uri(model_name): + """Resolve the S3 artifact URI of a built model, to filter recs by model.""" + container = getattr(Model.get(model_name=model_name), "primary_container", None) + if container is None: + return None + mds = getattr(container, "model_data_source", None) + if mds is not None: + s3 = getattr(mds, "s3_data_source", None) + if s3 is not None and getattr(s3, "s3_uri", None): + return s3.s3_uri + return getattr(container, "model_data_url", None) + + +def _delete_quietly(resource_factory, label): + """Best-effort delete; log and continue on any failure.""" + try: + resource_factory().delete() + logger.info("Deleted %s", label) + except Exception as exc: + logger.warning("Failed to delete %s: %s", label, exc) diff --git a/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_compare.py b/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_compare.py new file mode 100644 index 0000000000..92b186246a --- /dev/null +++ b/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_compare.py @@ -0,0 +1,163 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"). You +# may not use this file except in compliance with the License. A copy of +# the License is located at +# +# http://aws.amazon.com/apache2.0/ +# +# or in the "license" file accompanying this file. This file is +# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF +# ANY KIND, either express or implied. See the License for the specific +# language governing permissions and limitations under the License. +"""Unit tests for compare_benchmarks / BenchmarkComparison.""" +from __future__ import absolute_import + +import pytest + +from sagemaker.serve.ai_inference_recommender import ( + BenchmarkResult, + compare_benchmarks, +) +from sagemaker.serve.ai_inference_recommender.result import ( + BenchmarkMetrics, + BenchmarkSearchResult, +) + + +def _result(throughput=None, latency=None, s3="s3://b/out/"): + """A single-run BenchmarkResult with request_throughput / request_latency.""" + profile = {} + if throughput is not None: + profile["request_throughput"] = {"unit": "req/s", "avg": throughput, "p50": throughput} + if latency is not None: + profile["request_latency"] = {"unit": "ms", "avg": latency, "p50": latency} + return BenchmarkResult( + metrics=BenchmarkMetrics.from_profile_json(profile), + s3_output_location=s3, + ) + + +class TestCompareBenchmarks: + def test_requires_at_least_two(self): + with pytest.raises(ValueError, match="at least two"): + compare_benchmarks(_result(throughput=1.0)) + + def test_names_length_must_match(self): + with pytest.raises(ValueError, match="names has"): + compare_benchmarks(_result(throughput=1.0), _result(throughput=2.0), names=["only-one"]) + + def test_unknown_stat_rejected(self): + with pytest.raises(ValueError, match="stat must be one of"): + compare_benchmarks(_result(throughput=1.0), _result(throughput=2.0), stat="p42") + + def test_search_result_rejected(self): + search = BenchmarkResult( + metrics=BenchmarkMetrics.from_profile_json({}), + s3_output_location="s3://b/out/", + search=BenchmarkSearchResult(swept_dim="concurrency", winner=8), + ) + with pytest.raises(ValueError, match="search/sweep"): + compare_benchmarks(_result(throughput=1.0), search) + + def test_default_run_names(self): + cmp = compare_benchmarks(_result(throughput=1.0), _result(throughput=2.0)) + assert cmp.names == ["run1", "run2"] + + def test_custom_names_used(self): + cmp = compare_benchmarks( + _result(throughput=1.0), _result(throughput=2.0), names=["before", "after"] + ) + assert cmp.names == ["before", "after"] + + def test_throughput_increase_is_positive_delta(self): + # request_throughput is higher-is-better: 10 -> 15 is +50%. + cmp = compare_benchmarks(_result(throughput=10.0), _result(throughput=15.0)) + text = str(cmp) + assert "request_throughput" in text + assert "+50.0%" in text + + def test_latency_decrease_is_positive_delta(self): + # request_latency is lower-is-better: 100 -> 80 is a 20% improvement, + # reported as +20.0% (sign flipped so + is always better). + cmp = compare_benchmarks(_result(latency=100.0), _result(latency=80.0)) + text = str(cmp) + assert "request_latency" in text + assert "+20.0%" in text + + def test_latency_increase_is_negative_delta(self): + # 100 -> 120 latency is a regression: -20.0%. + cmp = compare_benchmarks(_result(latency=100.0), _result(latency=120.0)) + assert "-20.0%" in str(cmp) + + def test_table_has_a_column_per_run_and_delta(self): + cmp = compare_benchmarks( + _result(throughput=10.0), + _result(throughput=15.0), + _result(throughput=20.0), + names=["a", "b", "c"], + ) + text = str(cmp) + # value column per run + a delta column for each non-baseline run + assert "a" in text and "b" in text and "c" in text + assert "Δ% b" in text and "Δ% c" in text + # baseline note names the first run + assert "baseline: a" in text + + def test_stat_selects_percentile(self): + cmp = compare_benchmarks(_result(throughput=10.0), _result(throughput=15.0), stat="p50") + assert cmp.stat == "p50" + assert "+50.0%" in str(cmp) + + def test_missing_metric_renders_dash_not_crash(self): + # First run has throughput, second doesn't -> value '-' and delta '-'. + cmp = compare_benchmarks(_result(throughput=10.0), _result(latency=5.0)) + text = str(cmp) + assert "request_throughput" in text + assert "request_latency" in text + + +pd = pytest.importorskip("pandas") + + +class TestBenchmarkComparisonToDataFrame: + def test_columns_are_unit_runs_and_deltas(self): + cmp = compare_benchmarks( + _result(throughput=10.0), + _result(throughput=15.0), + _result(throughput=20.0), + names=["a", "b", "c"], + ) + df = cmp.to_dataframe() + assert list(df.columns) == ["unit", "a", "b", "c", "Δ% b", "Δ% c"] + assert df.index.name == "metric" + assert "request_throughput" in df.index + + def test_values_and_units_are_native(self): + cmp = compare_benchmarks( + _result(throughput=10.0), _result(throughput=15.0), names=["base", "cand"] + ) + row = cmp.to_dataframe().loc["request_throughput"] + assert row["base"] == 10.0 + assert row["cand"] == 15.0 + assert row["unit"] == "req/s" + + def test_delta_is_numeric_and_signed_for_direction(self): + # throughput up = better (+); latency up = worse (-). Deltas are numbers. + up = compare_benchmarks(_result(throughput=10.0), _result(throughput=15.0)) + assert up.to_dataframe().loc["request_throughput", "Δ% run2"] == pytest.approx(50.0) + + worse = compare_benchmarks(_result(latency=100.0), _result(latency=120.0)) + assert worse.to_dataframe().loc["request_latency", "Δ% run2"] == pytest.approx(-20.0) + + def test_missing_value_yields_nan_delta(self): + # Baseline has throughput, candidate does not -> value + delta are NaN. + cmp = compare_benchmarks(_result(throughput=10.0), _result(latency=5.0)) + df = cmp.to_dataframe() + assert pd.isna(df.loc["request_throughput", "run2"]) + assert pd.isna(df.loc["request_throughput", "Δ% run2"]) + + def test_stat_selects_percentile(self): + cmp = compare_benchmarks(_result(throughput=10.0), _result(throughput=15.0), stat="p50") + # p50 was set equal to avg in the builder; delta still +50%. + assert cmp.to_dataframe().loc["request_throughput", "Δ% run2"] == pytest.approx(50.0) diff --git a/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_listing.py b/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_listing.py new file mode 100644 index 0000000000..ea8ddf4d14 --- /dev/null +++ b/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_listing.py @@ -0,0 +1,162 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"). You +# may not use this file except in compliance with the License. A copy of +# the License is located at +# +# http://aws.amazon.com/apache2.0/ +# +# or in the "license" file accompanying this file. This file is +# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF +# ANY KIND, either express or implied. See the License for the specific +# language governing permissions and limitations under the License. +"""Unit tests for list_benchmarks / list_recommendations client-side filtering.""" +from __future__ import absolute_import + +from unittest.mock import patch + +import pytest + +from sagemaker.core.shapes.shapes import ( + AIBenchmarkEndpoint, + AIBenchmarkTarget, + AIModelSource, + AIModelSourceS3, + AIRecommendation, + AIRecommendationModelDetails, + AIRecommendationOutputResult, +) +from sagemaker.serve.ai_inference_recommender import ( + list_benchmarks, + list_recommendations, +) +from sagemaker.serve.ai_inference_recommender.jobs import BenchmarkJob, RecommendationJob + + +# Real BenchmarkJob / RecommendationJob instances (not SimpleNamespace): list_* +# reassigns each returned job's __class__ to the subclass — a no-op on a real +# instance, but a layout error on a plain namespace — so use the real classes. +# The nested fields the client-side filter reads are set directly on the +# instance. refresh() is class-patched to a no-op below (the stand-ins are +# pre-hydrated, so there is nothing to re-fetch). + + +@pytest.fixture(autouse=True) +def _no_op_refresh(): + """Stub refresh() so the client-side filter's hydration step is a no-op — + the test stand-ins already carry their nested fields.""" + with patch.object(BenchmarkJob, "refresh", lambda self: self), patch.object( + RecommendationJob, "refresh", lambda self: self + ): + yield + + +def _bench(name, endpoint_identifier=None): + job = BenchmarkJob(ai_benchmark_job_name=name) + job.benchmark_target = ( + AIBenchmarkTarget(endpoint=AIBenchmarkEndpoint(identifier=endpoint_identifier)) + if endpoint_identifier is not None + else None + ) + return job + + +def _rec(name, s3_uri=None, group=None, row_arns=()): + job = RecommendationJob(ai_recommendation_job_name=name) + job.model_source = ( + AIModelSource(s3=AIModelSourceS3(s3_uri=s3_uri)) if s3_uri else None + ) + job.output_config = ( + AIRecommendationOutputResult( + s3_output_location="s3://bucket/out/", + model_package_group_identifier=group, + ) + if group is not None + else None + ) + job.recommendations = [ + AIRecommendation(model_details=AIRecommendationModelDetails(model_package_arn=arn)) + for arn in row_arns + ] + return job + + +class TestListBenchmarks: + def test_no_filter_returns_all_native(self): + jobs = [_bench("b1"), _bench("b2")] + with patch.object(BenchmarkJob, "get_all", return_value=iter(jobs)): + out = list_benchmarks() + assert [j.ai_benchmark_job_name for j in out] == ["b1", "b2"] + + def test_endpoint_filter_matches_by_name(self): + jobs = [ + _bench("b1", endpoint_identifier="ep-A"), + _bench("b2", endpoint_identifier="ep-B"), + _bench("b3", endpoint_identifier=None), + ] + with patch.object(BenchmarkJob, "get_all", return_value=iter(jobs)): + out = list_benchmarks(endpoint="ep-A") + assert [j.ai_benchmark_job_name for j in out] == ["b1"] + + def test_endpoint_filter_matches_arn_suffix(self): + arn = "arn:aws:sagemaker:us-west-2:1:endpoint/ep-A" + jobs = [_bench("b1", endpoint_identifier=arn)] + with patch.object(BenchmarkJob, "get_all", return_value=iter(jobs)): + out = list_benchmarks(endpoint="ep-A") + assert [j.ai_benchmark_job_name for j in out] == ["b1"] + + def test_max_results_caps_output(self): + jobs = [_bench(f"b{i}") for i in range(10)] + with patch.object(BenchmarkJob, "get_all", return_value=iter(jobs)): + out = list_benchmarks(max_results=3) + assert len(out) == 3 + + def test_native_filters_forwarded(self): + with patch.object(BenchmarkJob, "get_all", return_value=iter([])) as get_all: + list_benchmarks(status="Completed", name_contains="qwen") + kwargs = get_all.call_args.kwargs + assert kwargs["status_equals"] == "Completed" + assert kwargs["name_contains"] == "qwen" + + +class TestListRecommendations: + def test_model_filter_matches_source_uri(self): + jobs = [ + _rec("r1", s3_uri="s3://bucket/model-a/"), + _rec("r2", s3_uri="s3://bucket/model-b/"), + ] + with patch.object(RecommendationJob, "get_all", return_value=iter(jobs)): + out = list_recommendations(model="s3://bucket/model-a/") + assert [j.ai_recommendation_job_name for j in out] == ["r1"] + + def test_model_filter_ignores_trailing_slash(self): + jobs = [_rec("r1", s3_uri="s3://bucket/model-a/")] + with patch.object(RecommendationJob, "get_all", return_value=iter(jobs)): + out = list_recommendations(model="s3://bucket/model-a") + assert [j.ai_recommendation_job_name for j in out] == ["r1"] + + def test_model_package_matches_output_group(self): + jobs = [ + _rec("r1", group="my-group"), + _rec("r2", group="other-group"), + ] + with patch.object(RecommendationJob, "get_all", return_value=iter(jobs)): + out = list_recommendations(model_package="my-group") + assert [j.ai_recommendation_job_name for j in out] == ["r1"] + + def test_model_package_matches_recommendation_row_arn(self): + arn = "arn:aws:sagemaker:us-west-2:1:model-package/g/1" + jobs = [_rec("r1", row_arns=(arn,)), _rec("r2", row_arns=())] + with patch.object(RecommendationJob, "get_all", return_value=iter(jobs)): + out = list_recommendations(model_package=arn) + assert [j.ai_recommendation_job_name for j in out] == ["r1"] + + def test_model_and_model_package_mutually_exclusive(self): + with pytest.raises(ValueError, match="only one of"): + list_recommendations(model="s3://x/", model_package="arn:y") + + def test_no_filter_returns_all(self): + jobs = [_rec("r1"), _rec("r2")] + with patch.object(RecommendationJob, "get_all", return_value=iter(jobs)): + out = list_recommendations() + assert [j.ai_recommendation_job_name for j in out] == ["r1", "r2"] diff --git a/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_model_builder_recommendations.py b/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_model_builder_recommendations.py index 9354b01c68..7bb5ce87a5 100644 --- a/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_model_builder_recommendations.py +++ b/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_model_builder_recommendations.py @@ -38,6 +38,14 @@ def mb_class(): return ModelBuilder +def _mock_session(): + """A mock SageMaker session whose ``sagemaker_config`` is a real dict, so the + full ``deploy()`` path (which validates that config) runs against a mock.""" + session = MagicMock() + session.sagemaker_config = {} + return session + + def _rec_row( spec_name=None, model_package_arn="arn:aws:sm:us-west-2:1:model-package/p/1", @@ -645,3 +653,62 @@ def test_multiple_matches_warn_and_use_first(self, mb_class, caplog): ) # First match (row A) is the one described/deployed. assert sm.describe_model_package.call_args.kwargs["ModelPackageName"] == "arn:.../p/A" + + +class TestDeployRecommendationRowObject: + """deploy(recommendation=) resolves a recommendation row to the + spec-name / index the deploy path already understands, instead of making + callers hand-copy a magic index.""" + + def _make_builder(self, mb_class): + mb = mb_class(sagemaker_session=_mock_session()) + mb._recommendation_job = SimpleNamespace( + recommendations=[ + _rec_row(spec_name="A", model_package_arn="arn:.../p/A"), + _rec_row(spec_name="B", model_package_arn="arn:.../p/B"), + ], + ai_recommendation_job_status="Completed", + ) + return mb + + def _row(self, mb, index): + # A real recommendation view row, as mb.recommendations[index] returns. + return mb.recommendations[index] + + def test_best_row_resolves_to_its_spec_name(self, mb_class): + mb = self._make_builder(mb_class) + with patch.object(mb_class, "_deploy_recommendation") as deploy_rec: + mb.deploy(recommendation=mb.recommendations.best) + # .best is row 0 (spec A); it should forward spec_name="A". + assert deploy_rec.call_args.kwargs["recommendation_spec_name"] == "A" + + def test_indexed_row_resolves_to_its_spec_name(self, mb_class): + mb = self._make_builder(mb_class) + with patch.object(mb_class, "_deploy_recommendation") as deploy_rec: + mb.deploy(recommendation=self._row(mb, 1)) + assert deploy_rec.call_args.kwargs["recommendation_spec_name"] == "B" + + def test_row_without_spec_name_falls_back_to_index(self, mb_class): + mb = mb_class(sagemaker_session=_mock_session()) + mb._recommendation_job = SimpleNamespace( + recommendations=[ + _rec_row(spec_name=None, model_package_arn="arn:.../p/0"), + _rec_row(spec_name=None, model_package_arn="arn:.../p/1"), + ], + ai_recommendation_job_status="Completed", + ) + with patch.object(mb_class, "_deploy_recommendation") as deploy_rec: + mb.deploy(recommendation=mb.recommendations[1]) + # No spec name -> fall back to the row's positional index. + assert deploy_rec.call_args.kwargs["recommendation_spec_name"] is None + assert deploy_rec.call_args.kwargs["recommendation_index"] == 1 + + def test_recommendation_conflicts_with_index(self, mb_class): + mb = self._make_builder(mb_class) + with pytest.raises(ValueError, match="only one of"): + mb.deploy(recommendation=mb.recommendations.best, recommendation_index=1) + + def test_recommendation_conflicts_with_spec_name(self, mb_class): + mb = self._make_builder(mb_class) + with pytest.raises(ValueError, match="only one of"): + mb.deploy(recommendation=mb.recommendations.best, recommendation_spec_name="B") diff --git a/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_recommendation_view_dataframe.py b/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_recommendation_view_dataframe.py new file mode 100644 index 0000000000..916f4428b7 --- /dev/null +++ b/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_recommendation_view_dataframe.py @@ -0,0 +1,119 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"). You +# may not use this file except in compliance with the License. A copy of +# the License is located at +# +# http://aws.amazon.com/apache2.0/ +# +# or in the "license" file accompanying this file. This file is +# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF +# ANY KIND, either express or implied. See the License for the specific +# language governing permissions and limitations under the License. +"""Unit tests for to_dataframe() on the recommendation views.""" +from __future__ import absolute_import + +from types import SimpleNamespace + +import pytest + +from sagemaker.serve.ai_inference_recommender._recommendation_view import ( + _RecommendationsView, + _RecommendationView, +) + +pd = pytest.importorskip("pandas") + + +def _perf(metric, stat, value, unit): + return SimpleNamespace(metric=metric, stat=stat, value=value, unit=unit) + + +def _rec_row(spec_name="A", instance_type="ml.g6.12xlarge", instance_count=1, perf=None): + return SimpleNamespace( + model_details=SimpleNamespace( + model_package_arn="arn:aws:sm:us-west-2:1:model-package/p/1", + inference_specification_name=spec_name, + ), + deployment_configuration=SimpleNamespace( + instance_type=instance_type, + instance_count=instance_count, + copy_count_per_instance=1, + image_uri=".../djl-inference:0.36.0-lmi25.0.0-cu130", + ), + expected_performance=perf or [], + ) + + +class TestRecommendationViewToDataFrame: + def test_columns_and_rows_match_expected_performance(self): + perf = [ + _perf("RequestThroughput", "avg", 3.8, "requests/sec"), + _perf("RequestLatency", "p50", 206.6, "ms"), + ] + view = _RecommendationView(_rec_row(perf=perf), index=0) + df = view.to_dataframe() + assert list(df.columns) == ["metric", "stat", "value", "unit"] + assert len(df) == 2 + assert set(df["metric"]) == {"RequestThroughput", "RequestLatency"} + assert df.iloc[0]["value"] == 3.8 # native float, not preformatted + + def test_empty_expected_performance_yields_empty_frame_with_schema(self): + df = _RecommendationView(_rec_row(perf=[]), index=0).to_dataframe() + assert len(df) == 0 + assert list(df.columns) == ["metric", "stat", "value", "unit"] + + +class TestRecommendationsViewToDataFrame: + def _view(self): + rows = [ + _rec_row( + spec_name="A", + instance_type="ml.g6.12xlarge", + perf=[ + _perf("RequestThroughput", "avg", 3.8, "requests/sec"), + _perf("RequestLatency", "p50", 206.6, "ms"), + _perf("RequestLatency", "p90", 257.9, "ms"), + ], + ), + _rec_row(spec_name="B", instance_type="ml.g6.2xlarge", perf=[]), + ] + return _RecommendationsView(_RecommendationView(r, index=i) for i, r in enumerate(rows)) + + def test_one_row_per_recommendation_indexed_by_idx(self): + df = self._view().to_dataframe() + assert df.index.name == "idx" + assert list(df.index) == [0, 1] + # idx is the index, not a duplicated column + assert "idx" not in df.columns + + def test_columns_match_printed_table(self): + df = self._view().to_dataframe() + assert list(df.columns) == [ + "spec_name", + "instance_type", + "instances", + "copies/inst", + "container", + "req/s", + "tok/s", + "lat_p50", + "lat_p90", + "lat_p99", + "ttft_p50", + "itl_p50", + ] + + def test_numeric_columns_are_native_and_missing_is_nan(self): + df = self._view().to_dataframe() + assert df.loc[0, "req/s"] == 3.8 + assert df.loc[0, "lat_p50"] == 206.6 + assert df.loc[0, "instance_type"] == "ml.g6.12xlarge" + assert df.loc[0, "container"] == "lmi25.0.0" + # row B has no expected_performance -> numeric metric is NaN, not "-" + assert pd.isna(df.loc[1, "req/s"]) + + def test_empty_view_yields_empty_frame_with_schema(self): + df = _RecommendationsView().to_dataframe() + assert len(df) == 0 + assert "instance_type" in df.columns diff --git a/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_result.py b/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_result.py index 1904845a59..699dbf6865 100644 --- a/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_result.py +++ b/sagemaker-serve/tests/unit/test_ai_inference_recommender/test_result.py @@ -73,9 +73,7 @@ def test_get_returns_known_metric(self): assert metrics.get("does_not_exist") is None def test_metric_raw_preserves_full_dict(self): - metric = BenchmarkMetric.from_dict( - "x", {"avg": 1, "unit": "s", "extra": "kept"} - ) + metric = BenchmarkMetric.from_dict("x", {"avg": 1, "unit": "s", "extra": "kept"}) assert metric.raw == {"avg": 1, "unit": "s", "extra": "kept"} @@ -169,6 +167,74 @@ def test_repr_is_concise_single_line(self): assert text.startswith("BenchmarkResult(") +pd = pytest.importorskip("pandas") + + +class TestToDataFrame: + """to_dataframe() on BenchmarkMetrics and BenchmarkResult.""" + + def _result(self) -> BenchmarkResult: + # SAMPLE_PROFILE plus an http_* metric, so ordering can be asserted. + profile = dict(SAMPLE_PROFILE) + profile["http_req_waiting"] = {"avg": 90.0, "p50": 80.0, "unit": "ms"} + return BenchmarkResult( + metrics=BenchmarkMetrics.from_profile_json(profile), + s3_output_location="s3://bucket/results/", + ) + + def test_metrics_dataframe_shape_and_columns(self): + df = BenchmarkMetrics.from_profile_json(SAMPLE_PROFILE).to_dataframe() + assert df.index.name == "metric" + assert list(df.columns) == [ + "unit", + "avg", + "min", + "max", + "p50", + "p90", + "p95", + "p99", + "stddev", + ] + # one row per parsed metric + assert set(df.index) == set(BenchmarkMetrics.from_profile_json(SAMPLE_PROFILE).all_metrics) + + def test_metrics_dataframe_carries_stats_the_text_table_drops(self): + # min/max are absent from the printed table but present in the frame. + df = BenchmarkMetrics.from_profile_json(SAMPLE_PROFILE).to_dataframe() + assert df.loc["request_throughput", "avg"] == 12.5 + assert df.loc["request_throughput", "min"] == 10.0 + assert df.loc["request_throughput", "max"] == 15.0 + # request_latency carries min/max/p99 that the printed table omits. + assert df.loc["request_latency", "min"] == 200.0 + assert df.loc["request_latency", "p99"] == 1450.0 + + def test_result_dataframe_orders_headline_first_http_last(self): + df = self._result().to_dataframe() + names = list(df.index) + # headline metric leads; http_* transport metric trails. + assert names[0] == "request_throughput" + assert names[-1] == "http_req_waiting" + + def test_result_and_metrics_frames_share_columns(self): + result = self._result() + assert list(result.to_dataframe().columns) == list(result.metrics.to_dataframe().columns) + + def test_search_result_dataframe_raises(self): + search = BenchmarkResult( + metrics=BenchmarkMetrics.from_profile_json({}), + s3_output_location="s3://b/s/", + search=BenchmarkSearchResult(swept_dim="concurrency", winner=8), + ) + with pytest.raises(ValueError, match="search/sweep"): + search.to_dataframe() + + def test_empty_metrics_yield_empty_frame_with_schema(self): + df = BenchmarkMetrics.from_profile_json({}).to_dataframe() + assert len(df) == 0 + assert "avg" in df.columns + + class TestBenchmarkResultMetadataFields: """Lock down endpoint / workload_config / tool_version on BenchmarkResult.""" @@ -191,9 +257,7 @@ def test_str_shows_endpoint_workload_and_tool_version(self): assert "0.6.0" in text def test_str_renders_dashes_when_metadata_missing(self): - text = str( - self._result(endpoint=None, workload_config=None, tool_version=None) - ) + text = str(self._result(endpoint=None, workload_config=None, tool_version=None)) assert "endpoint: -" in text assert "workload_config: -" in text assert "tool_version: -" in text @@ -237,10 +301,12 @@ def test_tool_version_pulled_from_profile_top_level(self): assert result.tool_version == "0.6.1" def test_tool_version_pulled_from_profile_metadata(self): - archive_bytes = self._archive_with({ - **SAMPLE_PROFILE, - "metadata": {"version": "0.6.2"}, - }) + archive_bytes = self._archive_with( + { + **SAMPLE_PROFILE, + "metadata": {"version": "0.6.2"}, + } + ) result = self._parse_archive(archive_bytes) assert result.tool_version == "0.6.2" @@ -282,9 +348,7 @@ class _SessionStub: def client(self, name): return s3_client - return BenchmarkResult.from_s3( - "s3://b/p/", session=_SessionStub() - ) + return BenchmarkResult.from_s3("s3://b/p/", session=_SessionStub()) # A concurrency search writes search_history.json at the artifact root. Schema