diff --git a/README.md b/README.md index 6fa8c74..d954c47 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,7 @@ cloud_predictor.fit( result = cloud_predictor.predict(test_data) # Real-time inference endpoint -cloud_predictor.deploy() -result = cloud_predictor.predict_real_time(test_data) -cloud_predictor.cleanup_deployment() +endpoint = cloud_predictor.deploy() +result = endpoint.predict(test_data) +endpoint.delete_endpoint() ``` diff --git a/docs/api/multimodal.rst b/docs/api/multimodal.rst deleted file mode 100644 index 4b760d1..0000000 --- a/docs/api/multimodal.rst +++ /dev/null @@ -1,10 +0,0 @@ -Multimodal -========== - -.. currentmodule:: autogluon.cloud - -.. autosummary:: - :toctree: . - :template: custom_class.rst - - MultiModalCloudPredictor diff --git a/docs/index.md b/docs/index.md index cbb6359..87e64a0 100644 --- a/docs/index.md +++ b/docs/index.md @@ -109,9 +109,9 @@ cloud_predictor.fit( result = cloud_predictor.predict(test_data) # Real-time inference endpoint -cloud_predictor.deploy() -result = cloud_predictor.predict_real_time(test_data) -cloud_predictor.cleanup_deployment() +endpoint = cloud_predictor.deploy() +result = endpoint.predict(test_data) +endpoint.delete_endpoint() ``` → [Full walkthrough](tutorials/predictor-tabular.md) @@ -142,9 +142,9 @@ cloud_predictor.fit( result = cloud_predictor.predict(data) # Real-time inference endpoint -cloud_predictor.deploy() -result = cloud_predictor.predict_real_time(data) -cloud_predictor.cleanup_deployment() +endpoint = cloud_predictor.deploy() +result = endpoint.predict(data) +endpoint.delete_endpoint() ``` → [Full walkthrough](tutorials/predictor-timeseries.md) @@ -173,7 +173,6 @@ hidden: Setup Tabular Time Series -Multimodal ``` ```{toctree} diff --git a/docs/tutorials/predictor-tabular.md b/docs/tutorials/predictor-tabular.md index fd743ea..a41e233 100644 --- a/docs/tutorials/predictor-tabular.md +++ b/docs/tutorials/predictor-tabular.md @@ -54,47 +54,45 @@ A rough guideline: if you need predictions less often than once an hour and can ### Real-time inference -Deploy the predictor as a SageMaker endpoint with {py:meth}`~autogluon.cloud.TabularCloudPredictor.deploy`: +Deploy the predictor as a SageMaker endpoint with {py:meth}`~autogluon.cloud.TabularCloudPredictor.deploy`, then send requests through the returned {py:class}`~autogluon.cloud.TabularEndpoint`: ```python -cloud_predictor.deploy( - instance_type="ml.m5.2xlarge", -) -``` - -Optionally, you can also attach to a deployed endpoint with {py:meth}`~autogluon.cloud.TabularCloudPredictor.attach_endpoint`: - -```python -cloud_predictor.attach_endpoint(endpoint="ENDPOINT_NAME") +endpoint = cloud_predictor.deploy(instance_type="ml.m5.2xlarge") # takes a few minutes ``` -Send requests to the endpoint with {py:meth}`~autogluon.cloud.TabularCloudPredictor.predict_real_time`, which returns a pandas Series of predictions: +{py:meth}`~autogluon.cloud.TabularEndpoint.predict` returns a pandas Series of predictions: ```python -result = cloud_predictor.predict_real_time("test.csv") # DataFrame, local path, or S3 URL +result = endpoint.predict("test.csv") # DataFrame, local path, or S3 URL # 0 dog # 1 cat # 2 cat # Name: label, dtype: object ``` -For class probabilities, use {py:meth}`~autogluon.cloud.TabularCloudPredictor.predict_proba_real_time`, which returns a DataFrame with one column per class: +For class probabilities, use {py:meth}`~autogluon.cloud.TabularEndpoint.predict_proba`. Like the batch version, it returns a `(predictions, probabilities)` tuple by default; pass `include_predict=False` to get the probabilities DataFrame alone: ```python -result = cloud_predictor.predict_proba_real_time("test.csv") +probabilities = endpoint.predict_proba("test.csv", include_predict=False) # dog cat # 0 0.682754 0.317246 # 1 0.195782 0.804218 # 2 0.372283 0.627717 ``` -Make sure you clean up the endpoint with {py:meth}`~autogluon.cloud.TabularCloudPredictor.cleanup_deployment`: +The endpoint stays active — and billed — until you delete it: ```python -cloud_predictor.cleanup_deployment() +endpoint.delete_endpoint() ``` -To check whether an endpoint is currently attached, call {py:meth}`~autogluon.cloud.TabularCloudPredictor.info` and look for the `endpoint` key in the returned dict. +To send requests to an endpoint that's already running (e.g. from a previous session), build a {py:class}`~autogluon.cloud.TabularEndpoint` directly from the endpoint name: + +```python +from autogluon.cloud import TabularEndpoint + +endpoint = TabularEndpoint(endpoint_name="ENDPOINT_NAME") +``` #### Invoke the endpoint without AutoGluon-Cloud The deployed endpoint is a normal SageMaker endpoint, and you can invoke it through other methods. For example, to invoke it with boto3 directly: diff --git a/docs/tutorials/predictor-timeseries.md b/docs/tutorials/predictor-timeseries.md index 6c004f7..cad4d8a 100644 --- a/docs/tutorials/predictor-timeseries.md +++ b/docs/tutorials/predictor-timeseries.md @@ -77,24 +77,16 @@ A rough guideline: if you need predictions less often than once an hour and can ### Real-time inference -Deploy the predictor as a SageMaker endpoint with {py:meth}`~autogluon.cloud.TimeSeriesCloudPredictor.deploy`: +Deploy the predictor as a SageMaker endpoint with {py:meth}`~autogluon.cloud.TimeSeriesCloudPredictor.deploy`, then send requests through the returned {py:class}`~autogluon.cloud.TimeSeriesEndpoint`: ```python -cloud_predictor.deploy( - instance_type="ml.m5.2xlarge", -) -``` - -Optionally, you can also attach to a deployed endpoint with {py:meth}`~autogluon.cloud.TimeSeriesCloudPredictor.attach_endpoint`: - -```python -cloud_predictor.attach_endpoint(endpoint="ENDPOINT_NAME") +endpoint = cloud_predictor.deploy(instance_type="ml.m5.2xlarge") # takes a few minutes ``` -Send requests to the endpoint with {py:meth}`~autogluon.cloud.TimeSeriesCloudPredictor.predict_real_time`. It takes the historical observations to forecast from, plus optional `known_covariates` (required when `known_covariates_names` was set at fit time) and `static_features`. The result is a DataFrame with one row per `(item_id, future timestamp)` pair and a column for each predicted quantile (plus the `mean`): +{py:meth}`~autogluon.cloud.TimeSeriesEndpoint.predict` takes the historical observations to forecast from, plus optional `known_covariates` (required when `known_covariates_names` was set at fit time) and `static_features`. The result is a DataFrame with one row per `(item_id, future timestamp)` pair and a column for each predicted quantile (plus the `mean`): ```python -forecasts = cloud_predictor.predict_real_time( +forecasts = endpoint.predict( "train.csv", # historical observations — forecasts start from the last timestamp per item known_covariates="known_covariates.csv", # required if known_covariates_names was set static_features="static_features.csv", # optional @@ -107,13 +99,19 @@ forecasts = cloud_predictor.predict_real_time( # ... ``` -Make sure you clean up the endpoint with {py:meth}`~autogluon.cloud.TimeSeriesCloudPredictor.cleanup_deployment`: +The prediction length, quantiles, and target are the ones set at fit time; requests that set them to different values are rejected. The endpoint stays active — and billed — until you delete it: ```python -cloud_predictor.cleanup_deployment() +endpoint.delete_endpoint() ``` -To check whether an endpoint is currently attached, call {py:meth}`~autogluon.cloud.TimeSeriesCloudPredictor.info` and look for the `endpoint` key in the returned dict. +To send requests to an endpoint that's already running (e.g. from a previous session), build a {py:class}`~autogluon.cloud.TimeSeriesEndpoint` directly from the endpoint name: + +```python +from autogluon.cloud import TimeSeriesEndpoint + +endpoint = TimeSeriesEndpoint(endpoint_name="ENDPOINT_NAME") +``` #### Invoke the endpoint without AutoGluon-Cloud The deployed endpoint is a normal SageMaker endpoint, so you can invoke it from any AWS SDK. The simplest payload is the historical observations as CSV — forecasts are generated starting from the last timestamp of each item: @@ -141,7 +139,7 @@ The CSV format only carries the historical observations. To pass `static_feature :animate: fade-in-slide-down :color: secondary -**Option 1: AutoGluon-Cloud's native `application/x-autogluon` envelope.** Each DataFrame is serialized as base64-encoded parquet and bundled in a single JSON object. This is what {py:meth}`~autogluon.cloud.TimeSeriesCloudPredictor.predict_real_time` sends under the hood: +**Option 1: AutoGluon-Cloud's native `application/x-autogluon` envelope.** Each DataFrame is serialized as base64-encoded parquet and bundled in a single JSON object. This is what {py:meth}`~autogluon.cloud.TimeSeriesEndpoint.predict` sends under the hood: ```python import base64 @@ -162,7 +160,7 @@ payload = { "data": df_to_b64(train_data), "known_covariates": df_to_b64(known_covariates), "static_features": df_to_b64(static_features), - "inference_kwargs": {}, # prediction_length / quantile_levels are baked in at fit time + "inference_kwargs": {}, # prediction_length / quantile_levels / target are fixed at fit time } client = boto3.client("sagemaker-runtime") diff --git a/pyproject.toml b/pyproject.toml index 8589169..030dbf8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,7 +47,7 @@ dependencies = [ "PyYAML~=6.0", "Pillow>=10.2,<13", "huggingface_hub>=0.20,<2", - "typing_extensions>=4.0,<5", + "typing_extensions>=4.5,<5", # 4.5 adds @deprecated (PEP 702) # CLI dependencies (autogluon-cloud command) "click>=8.0,<9", "rich>=13.0,<15", diff --git a/src/autogluon/cloud/__init__.py b/src/autogluon/cloud/__init__.py index ad6c638..87e3af5 100644 --- a/src/autogluon/cloud/__init__.py +++ b/src/autogluon/cloud/__init__.py @@ -3,6 +3,7 @@ from autogluon.common.utils.log_utils import _add_stream_handler from .cloud_setup import bootstrap, register, status, teardown +from .endpoint.multimodal_endpoint import MultiModalEndpoint from .endpoint.tabular_endpoint import TabularEndpoint from .endpoint.timeseries_endpoint import TimeSeriesEndpoint from .model.foundation_model import TabularFoundationModel, TimeSeriesFoundationModel @@ -13,6 +14,7 @@ __all__ = [ "MultiModalCloudPredictor", + "MultiModalEndpoint", "TabularCloudPredictor", "TabularEndpoint", "TabularFoundationModel", diff --git a/src/autogluon/cloud/backend/multimodal_sagemaker_backend.py b/src/autogluon/cloud/backend/multimodal_sagemaker_backend.py index 58f909e..4eac8dd 100644 --- a/src/autogluon/cloud/backend/multimodal_sagemaker_backend.py +++ b/src/autogluon/cloud/backend/multimodal_sagemaker_backend.py @@ -156,7 +156,7 @@ def predict( """ Predict using SageMaker batch transform. When minimizing latency isn't a concern, then the batch transform functionality may be easier, more scalable, and more appropriate. - If you want to minimize latency, use `predict_real_time()` instead. + If you want to minimize latency, deploy an endpoint with `deploy()` instead. To learn more: https://docs.aws.amazon.com/sagemaker/latest/dg/batch-transform.html Parameters @@ -197,7 +197,7 @@ def predict_proba( """ Predict proba using SageMaker batch transform. When minimizing latency isn't a concern, then the batch transform functionality may be easier, more scalable, and more appropriate. - If you want to minimize latency, use `predict_real_time()` instead. + If you want to minimize latency, deploy an endpoint with `deploy()` instead. To learn more: https://docs.aws.amazon.com/sagemaker/latest/dg/batch-transform.html Parameters diff --git a/src/autogluon/cloud/backend/sagemaker_backend.py b/src/autogluon/cloud/backend/sagemaker_backend.py index ad8ec81..31878c0 100644 --- a/src/autogluon/cloud/backend/sagemaker_backend.py +++ b/src/autogluon/cloud/backend/sagemaker_backend.py @@ -489,9 +489,6 @@ def deploy( :meth:`FoundationModel.cache_model_artifact`) to skip the round-trip. Ignored when ``predictor_path`` is None. """ - assert self.endpoint_name is None, ( - "There is an endpoint already attached. Either detach it with `detach` or clean it up with `cleanup_deployment`" - ) if inference_mode not in ("realtime", "serverless"): raise ValueError(f"Unsupported inference_mode={inference_mode!r}") overrides = check_override_keys(backend_overrides, DEPLOY_OVERRIDE_KEYS) @@ -811,7 +808,7 @@ def predict( """ Predict using SageMaker batch transform. When minimizing latency isn't a concern, then the batch transform functionality may be easier, more scalable, and more appropriate. - If you want to minimize latency, use `predict_real_time()` instead. + If you want to minimize latency, deploy an endpoint with `deploy()` instead. To learn more: https://docs.aws.amazon.com/sagemaker/latest/dg/batch-transform.html This method creates a SageMaker model with the trained predictor and runs a transform job with it. @@ -888,7 +885,7 @@ def predict_proba( """ Predict using SageMaker batch transform. When minimizing latency isn't a concern, then the batch transform functionality may be easier, more scalable, and more appropriate. - If you want to minimize latency, use `predict_real_time()` instead. + If you want to minimize latency, deploy an endpoint with `deploy()` instead. To learn more: https://docs.aws.amazon.com/sagemaker/latest/dg/batch-transform.html This method creates a SageMaker model with the trained predictor and runs a transform job with it. @@ -1190,7 +1187,8 @@ def _predict_real_time(self, test_data, accept, split_pred_proba=True, inference "The invocation of endpoint failed with Error Code 413. This is likely due to pay load size being too large." ) logger.warning( - "SageMaker endpoint could only take maximum 5MB. Please consider reduce test data size or use `predict()` instead." + "SageMaker endpoints accept at most 6 MB per request (4 MB for serverless endpoints). Please " + "send smaller batches, or use `predict()` for batch inference instead." ) raise e @@ -1269,7 +1267,7 @@ def _predict( # either a file to a dataframe, or a file to an image if is_image_file(test_data): logger.warning( - "Are you sure you want to do batch inference on a single image? You might want to try `deploy()` and `predict_real_time()` instead" + "Are you sure you want to do batch inference on a single image? You might want to try `deploy()` and `endpoint.predict()` instead" ) elif original_features is not None: # Loading is only needed for the column check below — skip it for predictors that don't track diff --git a/src/autogluon/cloud/backend/timeseries_sagemaker_backend.py b/src/autogluon/cloud/backend/timeseries_sagemaker_backend.py index 2ec301b..62b64e2 100644 --- a/src/autogluon/cloud/backend/timeseries_sagemaker_backend.py +++ b/src/autogluon/cloud/backend/timeseries_sagemaker_backend.py @@ -135,7 +135,7 @@ def predict( """ Predict using SageMaker batch transform. When minimizing latency isn't a concern, then the batch transform functionality may be easier, more scalable, and more appropriate. - If you want to minimize latency, use `predict_real_time()` instead. + If you want to minimize latency, deploy an endpoint with `deploy()` instead. To learn more: https://docs.aws.amazon.com/sagemaker/latest/dg/batch-transform.html Parameters diff --git a/src/autogluon/cloud/endpoint/endpoint.py b/src/autogluon/cloud/endpoint/endpoint.py new file mode 100644 index 0000000..bfabe4d --- /dev/null +++ b/src/autogluon/cloud/endpoint/endpoint.py @@ -0,0 +1,47 @@ +import logging +from typing import Any + +import boto3 + +from ..utils.aws_utils import setup_sagemaker_session +from ..utils.sagemaker_api import delete_endpoint + +logger = logging.getLogger(__name__) + + +class Endpoint: + """Base class for handles to SageMaker endpoints deployed through AutoGluon-Cloud.""" + + def __init__(self, endpoint_name: str, session: boto3.Session | None = None): + """ + Parameters + ---------- + endpoint_name: str + Name of an existing SageMaker endpoint deployed through AutoGluon-Cloud, either with ``deploy()`` of a + cloud predictor (e.g. :meth:`autogluon.cloud.TabularCloudPredictor.deploy`) or of a foundation model + (e.g. :meth:`autogluon.cloud.TabularFoundationModel.deploy`). + session: boto3.Session | None, default = None + ``boto3.Session`` used to invoke and delete the endpoint. If ``None``, the default ambient session is used. + """ + self._endpoint_name = endpoint_name + self._session = setup_sagemaker_session(boto_session=session) + + @property + def endpoint_name(self) -> str: + return self._endpoint_name + + @staticmethod + def _pop_as_pandas(inference_kwargs: dict[str, Any]) -> None: + # The serve scripts always pass as_pandas=True, so forwarding it would raise a duplicate-keyword TypeError. + if inference_kwargs.pop("as_pandas", True) is not True: + logger.warning("as_pandas must be True for real-time prediction; ignoring it.") + + def delete_endpoint(self) -> None: + """Delete the endpoint and its backing model + endpoint config. + + SageMaker API + ------------- + * :sm-api:`DeleteEndpoint`, :sm-api:`DeleteEndpointConfig` and :sm-api:`DeleteModel`: delete the endpoint and + the endpoint config and model created with it. + """ + delete_endpoint(self._endpoint_name, self._session) diff --git a/src/autogluon/cloud/endpoint/multimodal_endpoint.py b/src/autogluon/cloud/endpoint/multimodal_endpoint.py new file mode 100644 index 0000000..61ab8de --- /dev/null +++ b/src/autogluon/cloud/endpoint/multimodal_endpoint.py @@ -0,0 +1,129 @@ +from typing import Any + +import numpy as np +import pandas as pd + +from autogluon.common.loaders import load_pd + +from ..utils.deserializers import PandasDeserializer +from ..utils.sagemaker_api import invoke_endpoint +from ..utils.serializers import AutoGluonSerializationWrapper, MultiModalSerializer +from ..utils.utils import ( + convert_image_path_to_encoded_bytes_in_dataframe, + is_image_file, + read_image_bytes_and_encode, + split_pred_and_pred_proba, +) +from .endpoint import Endpoint + +DataInput = str | list[str] | pd.DataFrame +Prediction = pd.DataFrame | pd.Series + + +class MultiModalEndpoint(Endpoint): + """High-level handle for an AutoGluon-Cloud multimodal endpoint. + + Returned by :meth:`autogluon.cloud.MultiModalCloudPredictor.deploy`. Construct it directly to attach to an + existing endpoint by name. + """ + + @staticmethod + def _load_data(data: DataInput, image_column: str | None) -> tuple[pd.DataFrame | np.ndarray, str]: + if isinstance(data, str): + data = [data] if is_image_file(data) else load_pd.load(data) + if isinstance(data, list): + encoded_images = np.array([read_image_bytes_and_encode(image) for image in data], dtype="object") + return encoded_images, "application/x-autogluon-npy" + if image_column is not None: + data = convert_image_path_to_encoded_bytes_in_dataframe(dataframe=data, image_column=image_column) + return data, "application/x-autogluon-parquet" + + def _predict(self, data: DataInput, inference_kwargs: dict[str, Any]) -> tuple[pd.Series, Prediction]: + inference_kwargs = dict(inference_kwargs) + self._pop_as_pandas(inference_kwargs) + data, content_type = self._load_data(data, image_column=inference_kwargs.pop("image_column", None)) + raw = invoke_endpoint( + self._endpoint_name, + self._session, + AutoGluonSerializationWrapper(data=data, inference_kwargs=inference_kwargs), + serializer=MultiModalSerializer(), + deserializer=PandasDeserializer(), + # The serializer's content type is fixed, so the per-request content type is passed explicitly. + content_type=content_type, + accept="application/x-parquet", + ) + pred, pred_proba = split_pred_and_pred_proba(raw) + if pred_proba is None: + pred_proba = pred + return pred, pred_proba + + def predict(self, data: DataInput, **inference_kwargs: Any) -> pd.Series: + """Predict ``data`` with the deployed endpoint. + + This is intended for low-latency inference. For larger inputs, use + :meth:`autogluon.cloud.MultiModalCloudPredictor.predict` instead. + + Parameters + ---------- + data: str | list[str] | pd.DataFrame + Data to predict. One of: + + * a ``pd.DataFrame`` or a local path to a data file. + * a local path to a single image file, or a list of local paths to image files. + **inference_kwargs: Any + Additional args passed to the ``predict`` call of the AutoGluon predictor on the endpoint. If ``data`` has + an image column, pass ``image_column`` to name the column with absolute paths to local images; the images + are encoded and sent with the request. + + Returns + ------- + pd.Series + Predictions for ``data``. + + SageMaker API + ------------- + * :sm-runtime-api:`InvokeEndpoint`: sends the data to the endpoint and returns the predictions. The payload is + limited to 6 MB (4 MB for serverless endpoints). + """ + pred, _ = self._predict(data, inference_kwargs=inference_kwargs) + return pred + + def predict_proba( + self, + data: DataInput, + *, + include_predict: bool = True, + **inference_kwargs: Any, + ) -> tuple[pd.Series, Prediction] | Prediction: + """Predict class probabilities for ``data`` with the deployed endpoint. + + For regression, the probability result is identical to the prediction. + + Parameters + ---------- + data: str | list[str] | pd.DataFrame + Data to predict. One of: + + * a ``pd.DataFrame`` or a local path to a data file. + * a local path to a single image file, or a list of local paths to image files. + include_predict: bool, default = True + Whether to return the predictions along with the probabilities. Both are computed in the same request. + **inference_kwargs: Any + Additional args passed to the ``predict_proba`` call of the AutoGluon predictor on the endpoint. If ``data`` has + an image column, pass ``image_column`` to name the column with absolute paths to local images; the images + are encoded and sent with the request. + + Returns + ------- + tuple[pd.Series, pd.DataFrame | pd.Series] | pd.DataFrame | pd.Series + ``(prediction, predict_probability)`` if ``include_predict`` is True, otherwise ``predict_probability``. + + SageMaker API + ------------- + * :sm-runtime-api:`InvokeEndpoint`: sends the data to the endpoint and returns the predictions. The payload is + limited to 6 MB (4 MB for serverless endpoints). + """ + pred, pred_proba = self._predict(data, inference_kwargs=inference_kwargs) + if include_predict: + return pred, pred_proba + return pred_proba diff --git a/src/autogluon/cloud/endpoint/tabular_endpoint.py b/src/autogluon/cloud/endpoint/tabular_endpoint.py index d51707e..7c0ef5b 100644 --- a/src/autogluon/cloud/endpoint/tabular_endpoint.py +++ b/src/autogluon/cloud/endpoint/tabular_endpoint.py @@ -1,40 +1,32 @@ from pathlib import Path from typing import Any -import boto3 import pandas as pd from autogluon.common.loaders import load_pd -from ..utils.aws_utils import setup_sagemaker_session from ..utils.deserializers import PandasDeserializer -from ..utils.sagemaker_api import delete_endpoint, invoke_endpoint +from ..utils.sagemaker_api import invoke_endpoint from ..utils.serializers import AutoGluonSerializationWrapper, AutoGluonSerializer -from ..utils.utils import split_pred_and_pred_proba +from ..utils.utils import convert_image_path_to_encoded_bytes_in_dataframe, split_pred_and_pred_proba +from .endpoint import Endpoint DataInput = str | Path | pd.DataFrame Prediction = pd.DataFrame | pd.Series -class TabularEndpoint: - """High-level handle for an AutoGluon-Cloud tabular foundation-model endpoint.""" +class TabularEndpoint(Endpoint): + """High-level handle for an AutoGluon-Cloud tabular endpoint. - def __init__(self, endpoint_name: str, session: boto3.Session | None = None): - """ - Parameters - ---------- - endpoint_name: str - Name of an existing SageMaker endpoint deployed through - :meth:`autogluon.cloud.TabularFoundationModel.deploy`. - session: boto3.Session | None, default = None - ``boto3.Session`` used to invoke and delete the endpoint. If ``None``, the default ambient session is used. - """ - self._endpoint_name = endpoint_name - self._session = setup_sagemaker_session(boto_session=session) + Returned by :meth:`autogluon.cloud.TabularCloudPredictor.deploy` and + :meth:`autogluon.cloud.TabularFoundationModel.deploy`. Construct it directly to attach to an existing endpoint by + name. - @property - def endpoint_name(self) -> str: - return self._endpoint_name + * **Trained predictor endpoints** (:meth:`TabularCloudPredictor.deploy`) serve the fitted predictor: pass only + ``data``. + * **Foundation model endpoints** (:meth:`TabularFoundationModel.deploy`) fit a request-scoped predictor on the + labeled examples sent with each request: pass ``train_data`` and ``label`` along with ``data``. + """ @staticmethod def _load_data(data: DataInput) -> pd.DataFrame: @@ -45,24 +37,42 @@ def _load_data(data: DataInput) -> pd.DataFrame: def _predict( self, data: DataInput, - train_data: DataInput, - label: str, + train_data: DataInput | None = None, + label: str | None = None, inference_kwargs: dict[str, Any] | None = None, ) -> tuple[pd.Series, Prediction]: data = self._load_data(data) - train_data = self._load_data(train_data) - - if label not in train_data.columns: - raise ValueError(f"Label column {label!r} is not present in `train_data`.") - feature_columns = [column for column in train_data.columns if column != label] - missing_columns = [column for column in feature_columns if column not in data.columns] - if missing_columns: - raise ValueError(f"`data` is missing feature columns present in `train_data`: {missing_columns}.") + inference_kwargs = dict(inference_kwargs or {}) + self._pop_as_pandas(inference_kwargs) + image_column = inference_kwargs.pop("image_column", None) + + if (train_data is None) != (label is None): + raise ValueError( + "`train_data` and `label` must be passed together: pass both for foundation model endpoints, " + "and neither for endpoints deployed with `TabularCloudPredictor.deploy`." + ) + if train_data is not None: + if image_column is not None: + raise ValueError( + "`image_column` is only supported by endpoints deployed with `TabularCloudPredictor.deploy`; " + "foundation model endpoints do not support image features." + ) + train_data = self._load_data(train_data) + if label not in train_data.columns: + raise ValueError(f"Label column {label!r} is not present in `train_data`.") + feature_columns = [column for column in train_data.columns if column != label] + missing_columns = [column for column in feature_columns if column not in data.columns] + if missing_columns: + raise ValueError(f"`data` is missing feature columns present in `train_data`: {missing_columns}.") + inference_kwargs = {"label": label, **inference_kwargs} + + if image_column is not None: + data = convert_image_path_to_encoded_bytes_in_dataframe(data, image_column) payload = AutoGluonSerializationWrapper( data=data, train_data=train_data, - inference_kwargs={"label": label, **(inference_kwargs or {})}, + inference_kwargs=inference_kwargs, ) raw = invoke_endpoint( self._endpoint_name, @@ -80,19 +90,37 @@ def _predict( def predict( self, data: DataInput, - train_data: DataInput, - label: str, + train_data: DataInput | None = None, + label: str | None = None, **inference_kwargs: Any, ) -> pd.Series: - """Fit the foundation model on ``train_data`` and predict ``data``. + """Predict ``data`` with the deployed endpoint. - The request includes both ``train_data`` and ``data``. Use - :meth:`autogluon.cloud.TabularFoundationModel.predict` for inputs above the payload limit. + This is intended for low-latency inference. For inputs above the payload limit, use + :meth:`autogluon.cloud.TabularCloudPredictor.predict` or + :meth:`autogluon.cloud.TabularFoundationModel.predict` instead. + + Parameters + ---------- + data: str | pathlib.Path | pd.DataFrame + Rows to predict, as a ``pd.DataFrame`` or local/S3 path to a data file. + train_data: str | pathlib.Path | pd.DataFrame | None, default = None + Labeled examples the foundation model is fit on. Required for foundation model endpoints; must be + ``None`` for trained predictor endpoints. + label: str | None, default = None + Name of the label column in ``train_data``. Required if and only if ``train_data`` is passed. + **inference_kwargs: Any + Additional args passed to ``TabularPredictor.predict`` on the endpoint. + + Returns + ------- + pd.Series + Predictions for ``data``. SageMaker API ------------- * :sm-runtime-api:`InvokeEndpoint`: sends the data to the endpoint and returns the predictions. The payload is - limited to 6 MB. + limited to 6 MB (4 MB for serverless endpoints). """ pred, _ = self._predict( data=data, @@ -105,23 +133,41 @@ def predict( def predict_proba( self, data: DataInput, - train_data: DataInput, - label: str, + train_data: DataInput | None = None, + label: str | None = None, *, include_predict: bool = True, **inference_kwargs: Any, ) -> tuple[pd.Series, Prediction] | Prediction: - """Fit the foundation model and return class probabilities. + """Predict class probabilities for ``data`` with the deployed endpoint. - For regression, the probability result is identical to the prediction. + For regression, the probability result is identical to the prediction. For inputs above the payload limit, + use :meth:`autogluon.cloud.TabularCloudPredictor.predict_proba` or + :meth:`autogluon.cloud.TabularFoundationModel.predict_proba` instead. - The request includes both ``train_data`` and ``data``. Use - :meth:`autogluon.cloud.TabularFoundationModel.predict_proba` for inputs above the payload limit. + Parameters + ---------- + data: str | pathlib.Path | pd.DataFrame + Rows to predict, as a ``pd.DataFrame`` or local/S3 path to a data file. + train_data: str | pathlib.Path | pd.DataFrame | None, default = None + Labeled examples the foundation model is fit on. Required for foundation model endpoints; must be + ``None`` for trained predictor endpoints. + label: str | None, default = None + Name of the label column in ``train_data``. Required if and only if ``train_data`` is passed. + include_predict: bool, default = True + Whether to return the predictions along with the probabilities. Both are computed in the same request. + **inference_kwargs: Any + Additional args passed to ``TabularPredictor.predict_proba`` on the endpoint. + + Returns + ------- + tuple[pd.Series, pd.DataFrame | pd.Series] | pd.DataFrame | pd.Series + ``(prediction, predict_probability)`` if ``include_predict`` is True, otherwise ``predict_probability``. SageMaker API ------------- * :sm-runtime-api:`InvokeEndpoint`: sends the data to the endpoint and returns the predictions. The payload is - limited to 6 MB. + limited to 6 MB (4 MB for serverless endpoints). """ pred, pred_proba = self._predict( data=data, @@ -132,13 +178,3 @@ def predict_proba( if include_predict: return pred, pred_proba return pred_proba - - def delete_endpoint(self) -> None: - """Delete the endpoint and its backing model + endpoint config. - - SageMaker API - ------------- - * :sm-api:`DeleteEndpoint`, :sm-api:`DeleteEndpointConfig` and :sm-api:`DeleteModel`: delete the endpoint and - the endpoint config and model created with it. - """ - delete_endpoint(self._endpoint_name, self._session) diff --git a/src/autogluon/cloud/endpoint/timeseries_endpoint.py b/src/autogluon/cloud/endpoint/timeseries_endpoint.py index b400c01..c5980e9 100644 --- a/src/autogluon/cloud/endpoint/timeseries_endpoint.py +++ b/src/autogluon/cloud/endpoint/timeseries_endpoint.py @@ -1,52 +1,36 @@ from typing import Any -import boto3 import pandas as pd from autogluon.common.loaders import load_pd -from ..utils.aws_utils import setup_sagemaker_session from ..utils.deserializers import PandasDeserializer -from ..utils.sagemaker_api import delete_endpoint, invoke_endpoint +from ..utils.sagemaker_api import invoke_endpoint from ..utils.serializers import AutoGluonSerializationWrapper, AutoGluonSerializer +from .endpoint import Endpoint -class TimeSeriesEndpoint: - """High-level handle for an AutoGluon-Cloud time series inference endpoint. +class TimeSeriesEndpoint(Endpoint): + """High-level handle for an AutoGluon-Cloud time series endpoint. - Wraps a SageMaker endpoint with the AutoGluon-Cloud serializer/deserializer pair, providing a clean - :meth:`predict` interface. Use this to attach to an existing endpoint by name. To create a new endpoint, call - :meth:`autogluon.cloud.TimeSeriesFoundationModel.deploy`, which returns a :class:`TimeSeriesEndpoint` already - pointing at the new endpoint. - """ + Returned by :meth:`autogluon.cloud.TimeSeriesCloudPredictor.deploy` and + :meth:`autogluon.cloud.TimeSeriesFoundationModel.deploy`. Construct it directly to attach to an existing endpoint + by name. - def __init__(self, endpoint_name: str, session: boto3.Session | None = None): - """ - Parameters - ---------- - endpoint_name: str - Name of an existing SageMaker endpoint deployed via AutoGluon-Cloud (e.g. through - :meth:`autogluon.cloud.TimeSeriesFoundationModel.deploy`). The endpoint must understand the AutoGluon-Cloud - request payload format. - session: boto3.Session | None, default = None - ``boto3.Session`` used to invoke and delete the endpoint. If ``None``, the default ambient session is used. - """ - self._endpoint_name = endpoint_name - self._session = setup_sagemaker_session(boto_session=session) - - @property - def endpoint_name(self) -> str: - return self._endpoint_name + * **Trained predictor endpoints** (:meth:`TimeSeriesCloudPredictor.deploy`) use the ``prediction_length``, + ``quantile_levels``, and ``target`` set at fit time, and reject requests that set them to different values. + * **Foundation model endpoints** (:meth:`TimeSeriesFoundationModel.deploy`) read them from each request. + """ def predict( self, data: str | pd.DataFrame, known_covariates: str | pd.DataFrame | None = None, static_features: str | pd.DataFrame | None = None, - prediction_length: int = 1, - target: str = "target", - id_column: str = "item_id", - timestamp_column: str = "timestamp", + prediction_length: int | None = None, + target: str | None = None, + id_column: str | None = None, + timestamp_column: str | None = None, quantile_levels: list[float] | None = None, ) -> pd.DataFrame: """ @@ -62,17 +46,25 @@ def predict( Future values of the known covariates over the forecast horizon. static_features: str | pd.DataFrame | None, default = None Static (time-independent) features describing each individual time series. - prediction_length: int, default = 1 - Forecast horizon: how many time steps into the future the model should predict. - target: str, default = "target" - Name of the column that contains the target values to forecast. - id_column: str, default = "item_id" - Name of the column with the unique identifier of each time series (item). - timestamp_column: str, default = "timestamp" - Name of the column with the observation timestamps. + prediction_length: int | None, default = None + Forecast horizon: how many time steps into the future the model should predict. Defaults to 1 on + foundation model endpoints; trained predictor endpoints use the value set at fit time. + target: str | None, default = None + Name of the column that contains the target values to forecast. Defaults to ``"target"`` on foundation + model endpoints; trained predictor endpoints use the value set at fit time. + id_column: str | None, default = None + Name of the column with the unique identifier of each time series (item). Defaults to ``"item_id"`` on + foundation model endpoints; trained predictor endpoints use the column set at fit time. + timestamp_column: str | None, default = None + Name of the column with the observation timestamps. Defaults to ``"timestamp"`` on foundation model + endpoints; trained predictor endpoints use the column set at fit time. quantile_levels: list[float] | None, default = None List of increasing decimals between 0 and 1 specifying which quantiles to estimate. Defaults to - ``[0.1, 0.2, ..., 0.9]``. + ``[0.1, 0.2, ..., 0.9]`` on foundation model endpoints; trained predictor endpoints use the value set at + fit time. + + Trained predictor endpoints raise an error if ``prediction_length``, ``target``, or ``quantile_levels`` is + set to a value different from the one used at fit time. Returns ------- @@ -91,14 +83,18 @@ def predict( if isinstance(static_features, str): static_features = load_pd.load(static_features) + # Only send the args that were set: the endpoint falls back to its own defaults for the rest. inference_kwargs: dict[str, Any] = { - "prediction_length": prediction_length, - "target": target, - "id_column": id_column, - "timestamp_column": timestamp_column, + key: value + for key, value in { + "prediction_length": prediction_length, + "target": target, + "id_column": id_column, + "timestamp_column": timestamp_column, + "quantile_levels": quantile_levels, + }.items() + if value is not None } - if quantile_levels is not None: - inference_kwargs["quantile_levels"] = quantile_levels payload = AutoGluonSerializationWrapper( data=data, @@ -114,13 +110,3 @@ def predict( deserializer=PandasDeserializer(), accept="application/x-parquet", ) - - def delete_endpoint(self) -> None: - """Delete the endpoint and its backing model + endpoint config. - - SageMaker API - ------------- - * :sm-api:`DeleteEndpoint`, :sm-api:`DeleteEndpointConfig` and :sm-api:`DeleteModel`: delete the endpoint and - the endpoint config and model created with it. - """ - delete_endpoint(self._endpoint_name, self._session) diff --git a/src/autogluon/cloud/predictor/cloud_predictor.py b/src/autogluon/cloud/predictor/cloud_predictor.py index 108ccc0..cf7b2f0 100644 --- a/src/autogluon/cloud/predictor/cloud_predictor.py +++ b/src/autogluon/cloud/predictor/cloud_predictor.py @@ -4,13 +4,15 @@ import logging import os import tarfile +import warnings from abc import ABC, abstractmethod from datetime import datetime from pathlib import Path -from typing import Any, Literal +from typing import Any, Generic, Literal, TypeVar import boto3 import pandas as pd +from typing_extensions import deprecated from autogluon.common.loaders import load_pkl from autogluon.common.savers import save_pkl @@ -21,6 +23,7 @@ from ..backend.backend import Backend from ..backend.backend_factory import BackendFactory from ..backend.constant import SAGEMAKER +from ..endpoint.endpoint import Endpoint from ..utils.aws_utils import resolve_cloud_output_path from ..utils.constants import DEFAULT_FRAMEWORK_VERSION, DEFAULT_VOLUME_SIZE from ..utils.sagemaker_api import reject_legacy_kwargs @@ -28,10 +31,17 @@ logger = logging.getLogger(__name__) +EndpointT = TypeVar("EndpointT", bound=Endpoint) -class CloudPredictor(ABC): +# Marks methods as deprecated for IDEs and type checkers only (category=None); the methods emit their own +# FutureWarning at runtime, naming the concrete predictor and endpoint classes. +_DEPRECATED_REAL_TIME = "Deploy an endpoint with `deploy()` and call `predict()` / `predict_proba()` on it instead." + + +class CloudPredictor(ABC, Generic[EndpointT]): predictor_file_name = "CloudPredictor.pkl" backend_map = {} + _endpoint_cls: type[EndpointT] def __init__( self, @@ -396,10 +406,13 @@ def deploy( inference_mode: Literal["realtime", "serverless"] = "realtime", inference_config: dict[str, Any] | None = None, backend_overrides: dict[str, dict[str, Any]] | None = None, - ) -> None: + ) -> EndpointT: """ Deploy a predictor to an inference endpoint. + Returns a handle to the endpoint; call its ``predict()`` for low-latency inference, then + ``delete_endpoint()`` to tear it down. + Parameters ---------- predictor_path: str @@ -442,6 +455,11 @@ def deploy( built by AutoGluon-Cloud; lists and other non-dict values replace the generated ones. * Example: ``{"ProductionVariant": {"ModelDataDownloadTimeoutInSeconds": 1200}}`` + Returns + ------- + TabularEndpoint | TimeSeriesEndpoint | MultiModalEndpoint + Handle to the deployed endpoint, matching the predictor type. + SageMaker API ------------- * :sm-api:`CreateModel`: registers the model artifact and inference image as a SageMaker model. @@ -449,12 +467,13 @@ def deploy( count, or the serverless settings. * :sm-api:`CreateEndpoint`: launches the endpoint. - The endpoint is billed until :meth:`cleanup_deployment` deletes it. + The endpoint is billed until ``delete_endpoint()`` of the returned endpoint deletes it. """ if inference_mode == "serverless" and instance_type is not None: raise ValueError("`instance_type` must not be set when `inference_mode='serverless'`.") if instance_type is None and inference_mode == "realtime": instance_type = "ml.m5.2xlarge" + self._warn_if_endpoint_active() self.backend.deploy( predictor_path=predictor_path, endpoint_name=endpoint_name, @@ -468,29 +487,90 @@ def deploy( inference_config=inference_config, backend_overrides=backend_overrides, ) + return self._endpoint_cls( + endpoint_name=self.backend.endpoint_name, + session=self.backend.sagemaker_session.boto_session, + ) + def _warn_if_endpoint_active(self) -> None: + # The predictor only remembers the name of its last endpoint, which may since have been deleted through the + # returned endpoint handle, so ask SageMaker before warning. + previous_endpoint = self.backend.endpoint_name + if previous_endpoint is None: + return + try: + status = self.backend.sagemaker_session.sagemaker_client.describe_endpoint(EndpointName=previous_endpoint)[ + "EndpointStatus" + ] + except Exception: + return + if status not in ("Deleting", "Failed"): + logger.warning( + f"This predictor already deployed endpoint {previous_endpoint} (status: {status}). Deploying a new " + f"endpoint; {previous_endpoint} keeps running and incurring charges until you delete it with " + f"`{self._endpoint_cls.__name__}('{previous_endpoint}').delete_endpoint()`." + ) + + def _warn_deprecated(self, method: str, replacement: str, stacklevel: int = 3) -> None: + warnings.warn( + f"`{type(self).__name__}.{method}` is deprecated and will be removed in a future release. {replacement}", + FutureWarning, + stacklevel=stacklevel, + ) + + def _warn_deprecated_real_time(self, method: str) -> None: + endpoint_method = method.removesuffix("_real_time") + self._warn_deprecated( + method, + f"Use the endpoint returned by `deploy()` instead: `endpoint = predictor.deploy()`, then " + f"`endpoint.{endpoint_method}(...)`.", + stacklevel=4, + ) + + @deprecated("Construct the endpoint class directly from the endpoint name instead.", category=None) def attach_endpoint(self, endpoint: str) -> None: """ Attach the current CloudPredictor to an existing endpoint. + :meta private: + + .. deprecated:: + Construct the endpoint class directly to get a handle to an existing endpoint instead, e.g. + ``TabularEndpoint(endpoint_name)``. + Parameters ---------- endpoint: str Name of the endpoint being attached to. """ + self._warn_deprecated( + "attach_endpoint", + f"Use `{self._endpoint_cls.__name__}(endpoint_name)` to get a handle to an existing endpoint instead.", + ) self.backend.attach_endpoint(endpoint) + @deprecated("The endpoint returned by `deploy()` is independent of the predictor.", category=None) def detach_endpoint(self) -> str: """ Detach the current endpoint and return its name. + :meta private: + + .. deprecated:: + The endpoint returned by :meth:`deploy` is independent of the predictor, so there is nothing to detach. + Returns ------- str Name of the detached endpoint. Pass it to :meth:`attach_endpoint` to attach it again. """ + self._warn_deprecated( + "detach_endpoint", + "The endpoint returned by `deploy()` is independent of the predictor, so there is nothing to detach.", + ) return self.backend.detach_endpoint() + @deprecated(_DEPRECATED_REAL_TIME, category=None) def predict_real_time( self, test_data: str | pd.DataFrame, @@ -503,6 +583,11 @@ def predict_real_time( This is intended to provide a low latency inference. If you want to inference on a large dataset, use `predict()` instead. + :meta private: + + .. deprecated:: + Use ``predict()`` of the endpoint returned by :meth:`deploy` instead. + Parameters ---------- test_data: str | pd.DataFrame @@ -527,11 +612,13 @@ def predict_real_time( * :sm-runtime-api:`InvokeEndpoint`: sends the data to the endpoint and returns the predictions. The payload is limited to 6 MB (4 MB for serverless endpoints). """ + self._warn_deprecated_real_time("predict_real_time") self._validate_inference_kwargs(inference_kwargs=kwargs) return self.backend.predict_real_time( test_data=test_data, test_data_image_column=test_data_image_column, accept=accept, inference_kwargs=kwargs ) + @deprecated(_DEPRECATED_REAL_TIME, category=None) def predict_proba_real_time( self, test_data: str | pd.DataFrame, @@ -545,6 +632,11 @@ def predict_proba_real_time( If you want to inference on a large dataset, use `predict_proba()` instead. If your problem_type is regression, this functions identically to `predict_real_time`, returning the same output. + :meta private: + + .. deprecated:: + Use ``predict_proba(..., include_predict=False)`` of the endpoint returned by :meth:`deploy` instead. + Parameters ---------- test_data: str | pd.DataFrame @@ -569,9 +661,10 @@ def predict_proba_real_time( * :sm-runtime-api:`InvokeEndpoint`: sends the data to the endpoint and returns the predictions. The payload is limited to 6 MB (4 MB for serverless endpoints). """ + self._warn_deprecated_real_time("predict_proba_real_time") self._validate_inference_kwargs(inference_kwargs=kwargs) return self.backend.predict_proba_real_time( - test_data=test_data, test_data_image_column=test_data_image_column, accept=accept + test_data=test_data, test_data_image_column=test_data_image_column, accept=accept, inference_kwargs=kwargs ) @reject_legacy_kwargs @@ -592,7 +685,7 @@ def predict( """ Batch inference. When minimizing latency isn't a concern, then the batch transform functionality may be easier, more scalable, and more appropriate. - If you want to minimize latency, use `predict_real_time()` instead. + If you want to minimize latency, deploy an endpoint with `deploy()` instead. Parameters ---------- @@ -678,7 +771,7 @@ def predict_proba( """ Batch inference When minimizing latency isn't a concern, then the batch transform functionality may be easier, more scalable, and more appropriate. - If you want to minimize latency, use `predict_real_time()` instead. + If you want to minimize latency, deploy an endpoint with `deploy()` instead. Parameters ---------- @@ -780,15 +873,24 @@ def get_batch_inference_job_status(self, job_name: str | None = None) -> str: """ return self.backend.get_batch_inference_job_status(job_name) + @deprecated("Call `delete_endpoint()` on the endpoint returned by `deploy()` instead.", category=None) def cleanup_deployment(self) -> None: """ Delete the deployed endpoint and other artifacts + :meta private: + + .. deprecated:: + Use ``delete_endpoint()`` of the endpoint returned by :meth:`deploy` instead. + SageMaker API ------------- * :sm-api:`DeleteEndpoint`, :sm-api:`DeleteEndpointConfig` and :sm-api:`DeleteModel`: delete the endpoint and the endpoint config and model created with it. """ + self._warn_deprecated( + "cleanup_deployment", "Use `delete_endpoint()` of the endpoint returned by `deploy()` instead." + ) self.backend.cleanup_deployment() def _download_predictor(self, path, save_path): diff --git a/src/autogluon/cloud/predictor/multimodal_cloud_predictor.py b/src/autogluon/cloud/predictor/multimodal_cloud_predictor.py index 48976bc..72efc21 100644 --- a/src/autogluon/cloud/predictor/multimodal_cloud_predictor.py +++ b/src/autogluon/cloud/predictor/multimodal_cloud_predictor.py @@ -2,12 +2,13 @@ import warnings from ..backend.constant import MULTIMODL_SAGEMAKER, SAGEMAKER +from ..endpoint.multimodal_endpoint import MultiModalEndpoint from .cloud_predictor import CloudPredictor logger = logging.getLogger(__name__) -class MultiModalCloudPredictor(CloudPredictor): +class MultiModalCloudPredictor(CloudPredictor[MultiModalEndpoint]): """Train and deploy AutoGluon multimodal models (image, text, tabular) on Amazon SageMaker. Wraps :class:`autogluon.multimodal.MultiModalPredictor` (`docs `_) @@ -16,6 +17,7 @@ class MultiModalCloudPredictor(CloudPredictor): predictor_file_name = "MultiModalCloudPredictor.pkl" backend_map = {SAGEMAKER: MULTIMODL_SAGEMAKER} + _endpoint_cls = MultiModalEndpoint def __init__(self, *args, **kwargs) -> None: warnings.warn( diff --git a/src/autogluon/cloud/predictor/tabular_cloud_predictor.py b/src/autogluon/cloud/predictor/tabular_cloud_predictor.py index e03dbc9..7480e6f 100644 --- a/src/autogluon/cloud/predictor/tabular_cloud_predictor.py +++ b/src/autogluon/cloud/predictor/tabular_cloud_predictor.py @@ -7,6 +7,7 @@ import pandas as pd from ..backend.constant import SAGEMAKER, TABULAR_SAGEMAKER +from ..endpoint.tabular_endpoint import TabularEndpoint from ..utils.constants import DEFAULT_FRAMEWORK_VERSION, DEFAULT_VOLUME_SIZE from ..utils.sagemaker_api import reject_legacy_kwargs from ..utils.utils import split_pred_and_pred_proba @@ -15,7 +16,7 @@ logger = logging.getLogger(__name__) -class TabularCloudPredictor(CloudPredictor): +class TabularCloudPredictor(CloudPredictor[TabularEndpoint]): """Train and deploy AutoGluon tabular models (classification and regression) on Amazon SageMaker. Wraps :class:`autogluon.tabular.TabularPredictor` (`docs `_) @@ -24,6 +25,7 @@ class TabularCloudPredictor(CloudPredictor): predictor_file_name = "TabularCloudPredictor.pkl" backend_map = {SAGEMAKER: TABULAR_SAGEMAKER} + _endpoint_cls = TabularEndpoint @property def predictor_type(self): diff --git a/src/autogluon/cloud/predictor/timeseries_cloud_predictor.py b/src/autogluon/cloud/predictor/timeseries_cloud_predictor.py index 66b2680..7544021 100644 --- a/src/autogluon/cloud/predictor/timeseries_cloud_predictor.py +++ b/src/autogluon/cloud/predictor/timeseries_cloud_predictor.py @@ -5,16 +5,18 @@ from typing import Any import pandas as pd +from typing_extensions import deprecated from ..backend.constant import SAGEMAKER, TIMESERIES_SAGEMAKER +from ..endpoint.timeseries_endpoint import TimeSeriesEndpoint from ..utils.constants import DEFAULT_FRAMEWORK_VERSION, DEFAULT_VOLUME_SIZE from ..utils.sagemaker_api import reject_legacy_kwargs -from .cloud_predictor import CloudPredictor +from .cloud_predictor import _DEPRECATED_REAL_TIME, CloudPredictor logger = logging.getLogger(__name__) -class TimeSeriesCloudPredictor(CloudPredictor): +class TimeSeriesCloudPredictor(CloudPredictor[TimeSeriesEndpoint]): """Train and deploy AutoGluon time series forecasting models on Amazon SageMaker. Wraps :class:`autogluon.timeseries.TimeSeriesPredictor` (`docs `_) @@ -23,6 +25,7 @@ class TimeSeriesCloudPredictor(CloudPredictor): predictor_file_name = "TimeSeriesCloudPredictor.pkl" backend_map = {SAGEMAKER: TIMESERIES_SAGEMAKER} + _endpoint_cls = TimeSeriesEndpoint @property def predictor_type(self): @@ -167,6 +170,7 @@ def fit( return self + @deprecated(_DEPRECATED_REAL_TIME, category=None) def predict_real_time( self, data: str | pd.DataFrame, @@ -180,6 +184,11 @@ def predict_real_time( This is intended to provide a low latency inference. If you want to inference on a large dataset, use `predict()` instead. + :meta private: + + .. deprecated:: + Use ``predict()`` of the endpoint returned by :meth:`deploy` instead. + ``data`` must use the same ``id_column`` / ``timestamp_column`` names that were passed to ``fit()``. Parameters @@ -208,6 +217,7 @@ def predict_real_time( * :sm-runtime-api:`InvokeEndpoint`: sends the data to the endpoint and returns the predictions. The payload is limited to 6 MB (4 MB for serverless endpoints). """ + self._warn_deprecated_real_time("predict_real_time") return self.backend.predict_real_time( test_data=data, static_features=static_features, @@ -241,7 +251,7 @@ def predict( """ Predict using SageMaker batch transform. When minimizing latency isn't a concern, then the batch transform functionality may be easier, more scalable, and more appropriate. - If you want to minimize latency, use `predict_real_time()` instead. + If you want to minimize latency, deploy an endpoint with `deploy()` instead. To learn more: https://docs.aws.amazon.com/sagemaker/latest/dg/batch-transform.html ``data`` must use the same ``id_column`` / ``timestamp_column`` names that were passed to ``fit()``. diff --git a/src/autogluon/cloud/scripts/sagemaker_scripts/tabular_serve.py b/src/autogluon/cloud/scripts/sagemaker_scripts/tabular_serve.py index 13cd1b0..8e37181 100644 --- a/src/autogluon/cloud/scripts/sagemaker_scripts/tabular_serve.py +++ b/src/autogluon/cloud/scripts/sagemaker_scripts/tabular_serve.py @@ -63,6 +63,11 @@ def transform_fn(model, request_body, input_content_type, output_content_type="a raise ValueError(f"Unsupported x-autogluon payload version: {payload.get('version')}. Expected 1.") data = pd.read_parquet(BytesIO(base64.b64decode(payload["data"]))) inference_kwargs = payload.get("inference_kwargs", {}) + if payload.get("train_data") is not None or "label" in (inference_kwargs or {}): + raise ValueError( + "This endpoint serves a trained predictor; `train_data` and `label` are only used by foundation " + "model endpoints. Omit them from the request." + ) data = _align_columns(data, column_names) else: diff --git a/src/autogluon/cloud/scripts/sagemaker_scripts/timeseries_serve.py b/src/autogluon/cloud/scripts/sagemaker_scripts/timeseries_serve.py index 2f0d807..8a690ba 100644 --- a/src/autogluon/cloud/scripts/sagemaker_scripts/timeseries_serve.py +++ b/src/autogluon/cloud/scripts/sagemaker_scripts/timeseries_serve.py @@ -3,6 +3,7 @@ import os import shutil +import numpy as np from autogluon.timeseries import TimeSeriesPredictor from serving_utils.timeseries import parse_payload, render_response @@ -34,15 +35,42 @@ def model_fn(model_dir): return model +def _check_fit_time_args(model, inference_kwargs): + """Reject requests asking for a different prediction_length / quantile_levels / target than the predictor was + fit with. These are fixed at fit time, so silently ignoring them would return a different forecast than asked for. + """ + fit_time_values = { + "prediction_length": model.prediction_length, + "quantile_levels": sorted(model.quantile_levels), + "target": model.target, + } + for key, fit_time_value in fit_time_values.items(): + if key not in inference_kwargs: + continue + value = inference_kwargs[key] + if key == "quantile_levels": + if not isinstance(value, (list, tuple)): + raise ValueError(f"`quantile_levels` must be a list of floats, got {value!r}.") + value = sorted(value) + # Tolerance absorbs float32 / serialization round-off; the length check guards against broadcasting. + matches = len(value) == len(fit_time_value) and np.allclose(value, fit_time_value) + else: + matches = value == fit_time_value + if not matches: + raise ValueError( + f"This endpoint serves a predictor fit with {key}={fit_time_value!r}, but the request has " + f"{key}={value!r}. Omit `{key}` from the request, or fit a new predictor." + ) + + def transform_fn(model, request_body, input_content_type, output_content_type="application/json"): - # prediction_length / quantile_levels are baked into the predictor at fit time, so - # any "parameters" block in a JumpStart payload is parsed but not applied. - tsdf, known_covariates, _ = parse_payload( + tsdf, known_covariates, inference_kwargs = parse_payload( request_body, input_content_type, id_column=model._id_column, timestamp_column=model._timestamp_column, target_column=model.target, ) + _check_fit_time_args(model, inference_kwargs) predictions = model.predict(tsdf, known_covariates=known_covariates) return render_response(predictions, output_content_type) diff --git a/tests/conftest.py b/tests/conftest.py index b6ce640..ed60002 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -68,27 +68,23 @@ def assert_ag_cloud_tags(arn: str, *, module: str, model_id: str = None): assert tags.get("autogluon-cloud-model-id") == model_id, f"missing/wrong model-id tag on {arn}: {tags}" @staticmethod - def test_endpoint(cloud_predictor, test_data, inference_kwargs=None, **predict_real_time_kwargs): - if inference_kwargs is None: - inference_kwargs = {} + def test_endpoint(endpoint, test_data, **inference_kwargs): try: - pred = cloud_predictor.predict_real_time(test_data, **inference_kwargs, **predict_real_time_kwargs) + pred = endpoint.predict(test_data, **inference_kwargs) assert isinstance(pred, pd.Series) - pred_proba = cloud_predictor.predict_proba_real_time( - test_data, **inference_kwargs, **predict_real_time_kwargs - ) + pred_proba = endpoint.predict_proba(test_data, include_predict=False, **inference_kwargs) assert isinstance(pred_proba, pd.DataFrame) except Exception as e: - cloud_predictor.cleanup_deployment() # cleanup endpoint if test failed + endpoint.delete_endpoint() # cleanup endpoint if test failed raise e @staticmethod - def test_timeseries_endpoint(cloud_predictor, test_data, **predict_real_time_kwargs): + def test_timeseries_endpoint(endpoint, test_data, **predict_kwargs): try: - pred = cloud_predictor.predict_real_time(test_data, **predict_real_time_kwargs) + pred = endpoint.predict(test_data, **predict_kwargs) assert isinstance(pred, pd.DataFrame) except Exception as e: - cloud_predictor.cleanup_deployment() # cleanup endpoint if test failed + endpoint.delete_endpoint() # cleanup endpoint if test failed raise e diff --git a/tests/unittests/general/test_cloud_predictor_deploy.py b/tests/unittests/general/test_cloud_predictor_deploy.py new file mode 100644 index 0000000..b34644f --- /dev/null +++ b/tests/unittests/general/test_cloud_predictor_deploy.py @@ -0,0 +1,128 @@ +"""Unit tests for CloudPredictor.deploy() returning an endpoint and the deprecated endpoint methods.""" + +import logging +from unittest import mock + +import pandas as pd +import pytest + +from autogluon.cloud import ( + MultiModalCloudPredictor, + MultiModalEndpoint, + TabularCloudPredictor, + TabularEndpoint, + TimeSeriesCloudPredictor, + TimeSeriesEndpoint, +) + + +@pytest.fixture(autouse=True) +def _stub_aws(monkeypatch): + """Avoid touching AWS / config files during construction.""" + monkeypatch.setattr( + "autogluon.cloud.predictor.cloud_predictor.resolve_cloud_output_path", + lambda path, backend_name: path or "s3://stub/output", + ) + monkeypatch.setattr( + "autogluon.cloud.backend.backend_factory.BackendFactory.get_backend", + lambda **kwargs: mock.MagicMock(endpoint_name="ep"), + ) + monkeypatch.setattr("autogluon.cloud.endpoint.endpoint.setup_sagemaker_session", lambda boto_session: boto_session) + # autogluon.common sets propagate=False on the `autogluon` logger, which hides records from caplog. + monkeypatch.setattr(logging.getLogger("autogluon"), "propagate", True) + + +@pytest.mark.filterwarnings("ignore:AutoGluon Multimodal is on a deprecation path") +@pytest.mark.parametrize( + "predictor_cls, endpoint_cls", + [ + (TabularCloudPredictor, TabularEndpoint), + (TimeSeriesCloudPredictor, TimeSeriesEndpoint), + (MultiModalCloudPredictor, MultiModalEndpoint), + ], +) +def test_deploy_returns_endpoint_matching_predictor_type(tmp_path, predictor_cls, endpoint_cls): + predictor = predictor_cls(local_output_path=str(tmp_path)) + predictor.backend.sagemaker_session.boto_session = mock.sentinel.boto_session + + endpoint = predictor.deploy() + + predictor.backend.deploy.assert_called_once() + assert type(endpoint) is endpoint_cls + assert endpoint.endpoint_name == "ep" + assert endpoint._session is mock.sentinel.boto_session + + +@pytest.mark.parametrize( + "call", + [ + lambda p: p.predict_real_time(pd.DataFrame({"a": [1]})), + lambda p: p.predict_proba_real_time(pd.DataFrame({"a": [1]})), + lambda p: p.attach_endpoint("ep"), + lambda p: p.detach_endpoint(), + lambda p: p.cleanup_deployment(), + ], + ids=["predict_real_time", "predict_proba_real_time", "attach_endpoint", "detach_endpoint", "cleanup_deployment"], +) +def test_legacy_endpoint_methods_warn_and_delegate_to_backend(tmp_path, call, request): + predictor = TabularCloudPredictor(local_output_path=str(tmp_path)) + method = request.node.callspec.id + + with pytest.warns(FutureWarning, match=f"TabularCloudPredictor.{method}` is deprecated") as record: + call(predictor) + + assert record[0].filename == __file__ + getattr(predictor.backend, method).assert_called_once() + + +def test_timeseries_predict_real_time_warns(tmp_path): + predictor = TimeSeriesCloudPredictor(local_output_path=str(tmp_path)) + + with pytest.warns(FutureWarning, match="endpoint.predict") as record: + predictor.predict_real_time(pd.DataFrame({"a": [1]})) + + assert record[0].filename == __file__ + predictor.backend.predict_real_time.assert_called_once() + + +def _describe_endpoint_returning(predictor, status=None, error=None): + describe = predictor.backend.sagemaker_session.sagemaker_client.describe_endpoint + describe.return_value = {"EndpointStatus": status} + describe.side_effect = error + return describe + + +@pytest.mark.parametrize("status", ["InService", "Creating"]) +def test_deploy_warns_when_previous_endpoint_is_active(tmp_path, caplog, status): + predictor = TabularCloudPredictor(local_output_path=str(tmp_path)) + describe = _describe_endpoint_returning(predictor, status=status) + + predictor.deploy() + + describe.assert_called_once_with(EndpointName="ep") + assert "already deployed endpoint ep" in caplog.text + assert "TabularEndpoint('ep').delete_endpoint()" in caplog.text + + +@pytest.mark.parametrize( + "status, error", + [("Deleting", None), ("Failed", None), (None, Exception("Could not find endpoint"))], + ids=["deleting", "failed", "not_found"], +) +def test_deploy_does_not_warn_when_previous_endpoint_is_gone(tmp_path, caplog, status, error): + predictor = TabularCloudPredictor(local_output_path=str(tmp_path)) + _describe_endpoint_returning(predictor, status=status, error=error) + + predictor.deploy() + + assert "already deployed endpoint" not in caplog.text + predictor.backend.deploy.assert_called_once() + + +def test_first_deploy_does_not_check_endpoint(tmp_path): + predictor = TabularCloudPredictor(local_output_path=str(tmp_path)) + predictor.backend.endpoint_name = None + + predictor.deploy() + + predictor.backend.sagemaker_session.sagemaker_client.describe_endpoint.assert_not_called() diff --git a/tests/unittests/general/test_tabular_endpoint.py b/tests/unittests/general/test_tabular_endpoint.py index b4bf152..0f66686 100644 --- a/tests/unittests/general/test_tabular_endpoint.py +++ b/tests/unittests/general/test_tabular_endpoint.py @@ -77,9 +77,70 @@ def test_predict_validates_label_and_feature_columns(make_endpoint, invoke_endpo endpoint.predict(data=pd.DataFrame({"other": [1]}), train_data=train_data, label="label") +def test_predict_without_train_data_sends_only_data(make_endpoint, invoke_endpoint): + data = pd.DataFrame({"feature": [2, 3]}) + endpoint = make_endpoint(pd.DataFrame({"label": ["a", "b"], "a_proba": [0.8, 0.2], "b_proba": [0.2, 0.8]})) + + pred = endpoint.predict(data, model="LightGBM") + + assert pred.tolist() == ["a", "b"] + payload = invoke_endpoint.call_args.args[2] + pd.testing.assert_frame_equal(payload.data, data) + assert payload.train_data is None + assert payload.inference_kwargs == {"model": "LightGBM"} + + +def test_predict_requires_train_data_and_label_together(make_endpoint, invoke_endpoint): + endpoint = make_endpoint(pd.DataFrame()) + data = pd.DataFrame({"feature": [1]}) + + with pytest.raises(ValueError, match="must be passed together"): + endpoint.predict(data, train_data=pd.DataFrame({"feature": [0], "label": ["a"]})) + + with pytest.raises(ValueError, match="must be passed together"): + endpoint.predict(data, label="label") + + invoke_endpoint.assert_not_called() + + +def test_image_column_is_encoded_and_not_forwarded(make_endpoint, invoke_endpoint): + endpoint = make_endpoint(pd.DataFrame({"target": [1.5]})) + data = pd.DataFrame({"image": ["/abs/img.png"]}) + + with mock.patch(f"{TE}.convert_image_path_to_encoded_bytes_in_dataframe") as convert: + endpoint.predict(data, image_column="image") + + convert.assert_called_once_with(data, "image") + payload = invoke_endpoint.call_args.args[2] + assert payload.data is convert.return_value + assert payload.inference_kwargs == {} + + +def test_image_column_with_train_data_raises(make_endpoint, invoke_endpoint): + endpoint = make_endpoint(pd.DataFrame({"label": ["a"]})) + train_data = pd.DataFrame({"image": ["/abs/img.png"], "label": ["a"]}) + + with pytest.raises(ValueError, match="`image_column` is only supported"): + endpoint.predict(pd.DataFrame({"image": ["/abs/img.png"]}), train_data, "label", image_column="image") + + invoke_endpoint.assert_not_called() + + +@pytest.mark.parametrize( + "train_data, label", [(None, None), (pd.DataFrame({"feature": [0], "label": ["a"]}), "label")] +) +def test_as_pandas_is_not_forwarded(make_endpoint, invoke_endpoint, train_data, label): + # The serve scripts pass as_pandas=True themselves; forwarding it raises a duplicate-keyword TypeError. + endpoint = make_endpoint(pd.DataFrame({"label": ["a"], "a_proba": [1.0]})) + + endpoint.predict(pd.DataFrame({"feature": [1]}), train_data, label, as_pandas=True, model="LightGBM") + + assert "as_pandas" not in invoke_endpoint.call_args.args[2].inference_kwargs + + def test_delete_endpoint_removes_model_endpoint_and_config(make_endpoint, invoke_endpoint): endpoint = make_endpoint(pd.DataFrame()) - with mock.patch(f"{TE}.delete_endpoint") as delete_endpoint: + with mock.patch("autogluon.cloud.endpoint.endpoint.delete_endpoint") as delete_endpoint: endpoint.delete_endpoint() delete_endpoint.assert_called_once_with("tabular-fm-endpoint", mock.sentinel.session) diff --git a/tests/unittests/general/test_timeseries_endpoint.py b/tests/unittests/general/test_timeseries_endpoint.py new file mode 100644 index 0000000..8f0af2e --- /dev/null +++ b/tests/unittests/general/test_timeseries_endpoint.py @@ -0,0 +1,46 @@ +from unittest import mock + +import pandas as pd +import pytest + +from autogluon.cloud.endpoint.timeseries_endpoint import TimeSeriesEndpoint + + +@pytest.fixture +def invoke_endpoint(): + with mock.patch("autogluon.cloud.endpoint.timeseries_endpoint.invoke_endpoint") as invoke: + yield invoke + + +@pytest.fixture +def endpoint(): + endpoint = TimeSeriesEndpoint.__new__(TimeSeriesEndpoint) + endpoint._endpoint_name = "ts-endpoint" + endpoint._session = mock.sentinel.session + return endpoint + + +def test_predict_omits_unset_args_so_endpoint_defaults_apply(endpoint, invoke_endpoint): + # Trained predictor endpoints read id/timestamp columns from fit-time metadata unless the request overrides them. + endpoint.predict(pd.DataFrame({"id": [1], "ts": ["2020-01-01"], "target": [0.0]})) + + assert invoke_endpoint.call_args.args[2].inference_kwargs == {} + + +def test_predict_forwards_set_args(endpoint, invoke_endpoint): + endpoint.predict( + pd.DataFrame({"id": [1], "ts": ["2020-01-01"], "y": [0.0]}), + prediction_length=3, + target="y", + id_column="id", + timestamp_column="ts", + quantile_levels=[0.5], + ) + + assert invoke_endpoint.call_args.args[2].inference_kwargs == { + "prediction_length": 3, + "target": "y", + "id_column": "id", + "timestamp_column": "ts", + "quantile_levels": [0.5], + } diff --git a/tests/unittests/tabular/test_tabular.py b/tests/unittests/tabular/test_tabular.py index cf3c1b8..8cf6065 100644 --- a/tests/unittests/tabular/test_tabular.py +++ b/tests/unittests/tabular/test_tabular.py @@ -8,7 +8,7 @@ import pytest from botocore.exceptions import ClientError -from autogluon.cloud import TabularCloudPredictor +from autogluon.cloud import TabularCloudPredictor, TabularEndpoint from autogluon.cloud.model import TabularFoundationModel from autogluon.cloud.model.registry import FOUNDATION_MODEL_REGISTRY, get_model_config from autogluon.common.utils.s3_utils import s3_path_to_bucket_prefix @@ -61,7 +61,7 @@ def test_tabular_train(test_helper, framework_version, shared_training_job_name) def test_tabular_endpoint_lifecycle(test_helper, framework_version, shared_training_job_name): - """Deploy the shared predictor and exercise detach, attach, save, and load.""" + """Deploy the shared predictor, re-attach to the endpoint by name, and exercise the deprecated predictor methods.""" with tempfile.TemporaryDirectory() as temp_dir: os.chdir(temp_dir) _prepare_data(test_helper) @@ -73,19 +73,18 @@ def test_tabular_endpoint_lifecycle(test_helper, framework_version, shared_train test_name="endpoint-lifecycle", ) - predictor.deploy(framework_version=framework_version) - endpoint_arn = boto3.client("sagemaker").describe_endpoint(EndpointName=predictor.endpoint_name)["EndpointArn"] + endpoint = predictor.deploy(framework_version=framework_version) + endpoint_arn = boto3.client("sagemaker").describe_endpoint(EndpointName=endpoint.endpoint_name)["EndpointArn"] test_helper.assert_ag_cloud_tags(endpoint_arn, module="tabular") - test_helper.test_endpoint(predictor, _TEST_DATA, inference_kwargs=dict(model="LightGBM")) + test_helper.test_endpoint(endpoint, _TEST_DATA, model="LightGBM") - detached_endpoint = predictor.detach_endpoint() - predictor.attach_endpoint(detached_endpoint) - test_helper.test_endpoint(predictor, _TEST_DATA) + test_helper.test_endpoint(TabularEndpoint(endpoint.endpoint_name), _TEST_DATA) - predictor.save() - predictor = TabularCloudPredictor.load(predictor.local_output_path) - test_helper.test_endpoint(predictor, _TEST_DATA) - predictor.cleanup_deployment() + try: + with pytest.warns(FutureWarning, match="predict_real_time"): + assert isinstance(predictor.predict_real_time(_TEST_DATA), pd.Series) + finally: + endpoint.delete_endpoint() def test_tabular_batch_predict(test_helper, framework_version, shared_training_job_name): @@ -120,12 +119,12 @@ def test_tabular_deploy_trained_artifact(test_helper, framework_version, shared_ local_output_path="test_tabular_deploy_trained_artifact", ) - predictor.deploy( + endpoint = predictor.deploy( predictor_path=artifact_path, framework_version=framework_version, ) - test_helper.test_endpoint(predictor, _TEST_DATA) - predictor.cleanup_deployment() + test_helper.test_endpoint(endpoint, _TEST_DATA) + endpoint.delete_endpoint() def test_tabular_predict_trained_artifact(test_helper, framework_version, shared_training_job_name): @@ -321,12 +320,12 @@ def test_tabular_cloud_predictor_gpu_release(test_helper, framework_version): framework_version=framework_version, ) - predictor.deploy( + endpoint = predictor.deploy( instance_type="ml.g4dn.xlarge", framework_version=framework_version, ) - test_helper.test_endpoint(predictor, _TEST_DATA) - predictor.cleanup_deployment() + test_helper.test_endpoint(endpoint, _TEST_DATA) + endpoint.delete_endpoint() pred, pred_proba = predictor.predict_proba( _TEST_DATA, diff --git a/tests/unittests/timeseries/test_timeseries.py b/tests/unittests/timeseries/test_timeseries.py index 2ef2d50..4858646 100644 --- a/tests/unittests/timeseries/test_timeseries.py +++ b/tests/unittests/timeseries/test_timeseries.py @@ -12,7 +12,7 @@ import pytest from botocore.exceptions import ClientError -from autogluon.cloud import TimeSeriesCloudPredictor +from autogluon.cloud import TimeSeriesCloudPredictor, TimeSeriesEndpoint from autogluon.cloud.model import FoundationModel from autogluon.cloud.model.registry import FOUNDATION_MODEL_REGISTRY, get_model_config from autogluon.common.utils.s3_utils import s3_path_to_bucket_prefix @@ -65,7 +65,7 @@ def _predict_kwargs(framework_version: str, ds: dict) -> dict: } -def _predict_real_time_kwargs(ds: dict) -> dict: +def _endpoint_predict_kwargs(ds: dict) -> dict: return { "static_features": ds["static_features"], "known_covariates": ds["known_covariates"], @@ -107,7 +107,7 @@ def test_timeseries_train(test_helper, framework_version, shared_training_job_na def test_timeseries_endpoint_lifecycle(test_helper, framework_version, shared_training_job_name, retail_sales_dataset): - """Deploy the shared predictor and exercise detach, attach, save, and load.""" + """Deploy the shared predictor, re-attach to the endpoint by name, and exercise the deprecated predictor methods.""" ds = retail_sales_dataset with tempfile.TemporaryDirectory() as temp_dir: os.chdir(temp_dir) @@ -118,23 +118,21 @@ def test_timeseries_endpoint_lifecycle(test_helper, framework_version, shared_tr job_name=shared_training_job_name, test_name="endpoint-lifecycle", ) - predictor.deploy(framework_version=framework_version) - endpoint_arn = boto3.client("sagemaker").describe_endpoint(EndpointName=predictor.endpoint_name)["EndpointArn"] + endpoint = predictor.deploy(framework_version=framework_version) + endpoint_arn = boto3.client("sagemaker").describe_endpoint(EndpointName=endpoint.endpoint_name)["EndpointArn"] test_helper.assert_ag_cloud_tags(endpoint_arn, module="timeseries") + test_helper.test_timeseries_endpoint(endpoint, ds["train_data"], **_endpoint_predict_kwargs(ds)) + test_helper.test_timeseries_endpoint( - predictor, - ds["train_data"], - **_predict_real_time_kwargs(ds), + TimeSeriesEndpoint(endpoint.endpoint_name), ds["train_data"], **_endpoint_predict_kwargs(ds) ) - detached_endpoint = predictor.detach_endpoint() - predictor.attach_endpoint(detached_endpoint) - test_helper.test_timeseries_endpoint(predictor, ds["train_data"], **_predict_real_time_kwargs(ds)) - - predictor.save() - predictor = TimeSeriesCloudPredictor.load(predictor.local_output_path) - test_helper.test_timeseries_endpoint(predictor, ds["train_data"], **_predict_real_time_kwargs(ds)) - predictor.cleanup_deployment() + try: + with pytest.warns(FutureWarning, match="predict_real_time"): + predictions = predictor.predict_real_time(ds["train_data"], **_endpoint_predict_kwargs(ds)) + assert isinstance(predictions, pd.DataFrame) + finally: + endpoint.delete_endpoint() def test_timeseries_batch_predict(test_helper, framework_version, shared_training_job_name, retail_sales_dataset): @@ -172,9 +170,9 @@ def test_timeseries_deploy_trained_artifact( local_output_path="test_timeseries_deploy_trained_artifact", ) - predictor.deploy(predictor_path=artifact_path, framework_version=framework_version) - test_helper.test_timeseries_endpoint(predictor, ds["train_data"], **_predict_real_time_kwargs(ds)) - predictor.cleanup_deployment() + endpoint = predictor.deploy(predictor_path=artifact_path, framework_version=framework_version) + test_helper.test_timeseries_endpoint(endpoint, ds["train_data"], **_endpoint_predict_kwargs(ds)) + endpoint.delete_endpoint() def test_timeseries_predict_trained_artifact( @@ -476,7 +474,7 @@ def test_timeseries_endpoint_payload_formats(test_helper, framework_version, pla predictor_fit_args=dict(presets="medium_quality", time_limit=60), framework_version=framework_version, ) - cloud_predictor.deploy(framework_version=framework_version) + endpoint = cloud_predictor.deploy(framework_version=framework_version) try: format_pairs = list( itertools.product( @@ -491,14 +489,14 @@ def test_timeseries_endpoint_payload_formats(test_helper, framework_version, pla ) ) _exercise_endpoint( - cloud_predictor.endpoint_name, + endpoint.endpoint_name, bodies, format_pairs, expected_item_ids=expected_item_ids, prediction_length=_PLAIN_PREDICTION_LENGTH, ) finally: - cloud_predictor.cleanup_deployment() + endpoint.delete_endpoint() @pytest.mark.parametrize( @@ -584,12 +582,12 @@ def test_timeseries_cloud_predictor_gpu_release(test_helper, framework_version, framework_version=framework_version, ) - predictor.deploy( + endpoint = predictor.deploy( instance_type="ml.g4dn.xlarge", framework_version=framework_version, ) - test_helper.test_timeseries_endpoint(predictor, ds["train_data"], **_predict_real_time_kwargs(ds)) - predictor.cleanup_deployment() + test_helper.test_timeseries_endpoint(endpoint, ds["train_data"], **_endpoint_predict_kwargs(ds)) + endpoint.delete_endpoint() predictions = predictor.predict( ds["train_data"],