Replace the SageMaker Python SDK with direct boto3 calls

pyproject.toml

@@ -43,8 +43,6 @@ dependencies = [
     # TransformAmiVersion was added to the SageMaker service model in botocore 1.37.23.
     "boto3>=1.37.23,<2",
     "packaging>=23.0,<27",
-    # Supports forwarding inference_ami_version from Model.deploy.
-    "sagemaker>=2.240.0,<3",
     "pyarrow>=19.0.1,<25",  # lower bound to avoid https://github.com/apache/arrow/issues/45283
     "PyYAML~=6.0",
     "Pillow>=10.2,<13",

src/autogluon/cloud/__init__.py

@@ -1,15 +1,4 @@
 import logging
-import os
-
-os.environ.setdefault("SAGEMAKER_SUPPRESS_V2_WARNING", "1")
-import sagemaker
-from packaging.version import Version
-
-if Version(sagemaker.__version__) >= Version("3.0"):
-    raise ImportError(
-        f"SageMaker SDK >= 3.0 is currently not supported (found {sagemaker.__version__}). "
-        "Please downgrade: pip install -U 'sagemaker<3'"
-    )
 
 from autogluon.common.utils.log_utils import _add_stream_handler
 

src/autogluon/cloud/backend/backend.py

@@ -7,8 +7,6 @@
 
 import pandas as pd
 
-from ..endpoint.endpoint import Endpoint
-
 
 def dumps_ag_args(config: Dict[str, Any]) -> str:
     """Serialize the remote-training config to JSON, raising a user-facing error on failure.
@@ -68,12 +66,7 @@ def initialize(
         self.predictor_type = predictor_type
         self.resource_prefix = resource_prefix or f"ag-cloud-{predictor_type}"
         self.original_features = None
-        self.endpoint: Optional[Endpoint] = None
-
-    @abstractmethod
-    def parse_backend_fit_kwargs(self, kwargs: Dict) -> Dict[str, Any]:
-        """Parse backend specific kwargs and get them ready to be sent to fit call"""
-        raise NotImplementedError
+        self.endpoint_name: Optional[str] = None
 
     @abstractmethod
     def attach_job(self, job_name: str) -> None:
@@ -139,11 +132,6 @@ def fit(self, **kwargs) -> None:
         """Fit AG on the backend"""
         raise NotImplementedError
 
-    @abstractmethod
-    def parse_backend_deploy_kwargs(self, kwargs: Dict) -> Dict[str, Any]:
-        """Parse backend specific kwargs and get them ready to be sent to deploy call"""
-        raise NotImplementedError
-
     @abstractmethod
     def deploy(self, **kwargs) -> None:
         """Deploy and endpoint"""
@@ -155,13 +143,13 @@ def cleanup_deployment(self, **kwargs) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def attach_endpoint(self, endpoint: Endpoint) -> None:
-        """Attach the backend to an existing endpoint"""
+    def attach_endpoint(self, endpoint: str) -> None:
+        """Attach the backend to an existing endpoint by name"""
         raise NotImplementedError
 
     @abstractmethod
-    def detach_endpoint(self) -> Endpoint:
-        """Detach the current endpoint and return it"""
+    def detach_endpoint(self) -> str:
+        """Detach the current endpoint and return its name"""
         raise NotImplementedError
 
     @abstractmethod
@@ -174,11 +162,6 @@ def predict_proba_real_time(self, test_data: Union[str, pd.DataFrame], **kwargs)
         """Realtime prediction probability with the endpoint"""
         raise NotImplementedError
 
-    @abstractmethod
-    def parse_backend_predict_kwargs(self, kwargs: Dict) -> Dict[str, Any]:
-        """Parse backend specific kwargs and get them ready to be sent to predict call"""
-        raise NotImplementedError
-
     @abstractmethod
     def get_batch_inference_job_info(self, job_name: Optional[str] = None) -> Dict[str, Any]:
         """

src/autogluon/cloud/backend/multimodal_sagemaker_backend.py

@@ -1,25 +1,24 @@
-import copy
 import os
 from typing import Any, Dict, Optional, Tuple, Union
 
 import pandas as pd
-from sagemaker import Predictor
 
 from autogluon.common.loaders import load_pd
 
-from ..utils.ag_sagemaker import AutoGluonMultiModalRealtimePredictor
+from ..utils.serializers import MultiModalSerializer
 from ..utils.utils import convert_image_path_to_encoded_bytes_in_dataframe, is_image_file, read_image_bytes_and_encode
 from .constant import MULTIMODL_SAGEMAKER
 from .sagemaker_backend import SagemakerBackend
 
 
 class MultiModalSagemakerBackend(SagemakerBackend):
     name = MULTIMODL_SAGEMAKER
+    # Images are sent one file per request.
+    _IMAGE_BATCH_ARGS = dict(content_type="application/x-image", split_type=None, batch_strategy="SingleRecord")
 
-    @property
-    def _realtime_predictor_cls(self) -> Predictor:
-        """Class used for realtime endpoint"""
-        return AutoGluonMultiModalRealtimePredictor
+    def _realtime_serializer(self):
+        """Serializer used for realtime endpoint requests"""
+        return MultiModalSerializer()
 
     def _load_predict_real_time_test_data(
         self, test_data: Union[str, pd.DataFrame], test_data_image_column: str
@@ -87,10 +86,9 @@ def predict_real_time(
         test_data, content_type = self._load_predict_real_time_test_data(
             test_data=test_data, test_data_image_column=test_data_image_column
         )
-        # Providing content type here because sagemaker serializer doesn't support change content type dynamically.
-        # Pass to `endpoint.predict()` call as `initial_args` instead
+        # The serializer's content type is fixed, so the per-request content type is passed explicitly.
         pred, _ = self._predict_real_time(
-            test_data=test_data, accept=accept, inference_kwargs=inference_kwargs, ContentType=content_type
+            test_data=test_data, accept=accept, inference_kwargs=inference_kwargs, content_type=content_type
         )
 
         return pred
@@ -139,10 +137,9 @@ def predict_proba_real_time(
         test_data, content_type = self._load_predict_real_time_test_data(
             test_data=test_data, test_data_image_column=test_data_image_column
         )
-        # Providing content type here because sagemaker serializer doesn't support change content type dynamically.
-        # Pass to `endpoint.predict()` call as `initial_args` instead
+        # The serializer's content type is fixed, so the per-request content type is passed explicitly.
         pred, proba = self._predict_real_time(
-            test_data=test_data, accept=accept, inference_kwargs=inference_kwargs, ContentType=content_type
+            test_data=test_data, accept=accept, inference_kwargs=inference_kwargs, content_type=content_type
         )
 
         if proba is None:
@@ -161,8 +158,6 @@ def predict(
         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.
         To learn more: https://docs.aws.amazon.com/sagemaker/latest/dg/batch-transform.html
-        This method would first create a AutoGluonSagemakerInferenceModel with the trained predictor,
-        then create a transformer with it, and call transform in the end.
 
         Parameters
         ----------
@@ -182,10 +177,10 @@ def predict(
         image_modality_only = self._check_image_modality_only(test_data)
 
         if image_modality_only:
-            processed_args = self._prepare_image_predict_args(**kwargs)
-            kwargs["transformer_kwargs"] = processed_args["transformer_kwargs"]
-            kwargs["transform_kwargs"] = processed_args["transform_kwargs"]
-            return super().predict(test_data, test_data_image_column=None, **kwargs)
+            pred, _ = self._predict(
+                test_data, original_features=self.original_features, **kwargs, **self._IMAGE_BATCH_ARGS
+            )
+            return pred
         else:
             return super().predict(
                 test_data,
@@ -204,8 +199,6 @@ def predict_proba(
         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.
         To learn more: https://docs.aws.amazon.com/sagemaker/latest/dg/batch-transform.html
-        This method would first create a AutoGluonSagemakerInferenceModel with the trained predictor,
-        then create a transformer with it, and call transform in the end.
 
         Parameters
         ----------
@@ -225,33 +218,18 @@ def predict_proba(
         image_modality_only = self._check_image_modality_only(test_data)
 
         if image_modality_only:
-            processed_args = self._prepare_image_predict_args(**kwargs)
-            kwargs["transformer_kwargs"] = processed_args["transformer_kwargs"]
-            kwargs["transform_kwargs"] = processed_args["transform_kwargs"]
-            return super().predict_proba(test_data, test_data_image_column=None, **kwargs)
+            include_predict = kwargs.pop("include_predict", True)
+            pred, pred_proba = self._predict(
+                test_data, original_features=self.original_features, **kwargs, **self._IMAGE_BATCH_ARGS
+            )
+            return (pred, pred_proba) if include_predict else pred_proba
         else:
             return super().predict_proba(
                 test_data,
                 test_data_image_column=test_data_image_column,
                 **kwargs,
             )
 
-    def _prepare_image_predict_args(self, **predict_kwargs):
-        split_type = None
-        content_type = "application/x-image"
-        predict_kwargs = copy.deepcopy(predict_kwargs)
-        transformer_kwargs = predict_kwargs.pop("transformer_kwargs", {})
-        if transformer_kwargs is None:
-            transformer_kwargs = {}
-        transformer_kwargs["strategy"] = "SingleRecord"
-        transform_kwargs = predict_kwargs.pop("transofrm_kwargs", {})
-        if transform_kwargs is None:
-            transform_kwargs = {}
-        transform_kwargs["split_type"] = split_type
-        transform_kwargs["content_type"] = content_type
-
-        return {"transformer_kwargs": transformer_kwargs, "transform_kwargs": transform_kwargs}
-
     def _check_image_modality_only(self, test_data):
         image_modality_only = False
         if isinstance(test_data, str):