feat: implement prediction service

.github/workflows/python-app.yml

@@ -44,7 +44,7 @@ jobs:
 
     strategy:
       matrix:
-        python-version: [3.7, 3.8]
+        python-version: [3.7]
     env:
       GITHUB_ACTION: true
 

colearn/ml_interface.py

@@ -20,32 +20,7 @@
 from typing import Any, Optional
 
 import onnx
-import onnxmltools
-import sklearn
-import tensorflow as tf
-import torch
 from pydantic import BaseModel
-from tensorflow import keras
-
-model_classes_keras = (tf.keras.Model, keras.Model, tf.estimator.Estimator)
-model_classes_scipy = (torch.nn.Module)
-model_classes_sklearn = (sklearn.base.ClassifierMixin)
-
-
-def convert_model_to_onnx(model: Any):
-    """
-    Helper function to convert a ML model to onnx format
-    """
-    if isinstance(model, model_classes_keras):
-        return onnxmltools.convert_keras(model)
-    if isinstance(model, model_classes_sklearn):
-        return onnxmltools.convert_sklearn(model)
-    if 'xgboost' in model.__repr__():
-        return onnxmltools.convert_sklearn(model)
-    if isinstance(model, model_classes_scipy):
-        raise Exception("Pytorch models not yet supported to onnx")
-    else:
-        raise Exception("Attempt to convert unsupported model to onnx: {model}")
 
 
 class DiffPrivBudget(BaseModel):
@@ -94,6 +69,17 @@ class ColearnModel(BaseModel):
     model: Optional[Any]
 
 
+class PredictionRequest(BaseModel):
+    name: str
+    input_data: Any
+    pred_data_loader_key: Optional[Any]
+
+
+class Prediction(BaseModel):
+    name: str
+    prediction_data: Any
+
+
 def deser_model(model: Any) -> onnx.ModelProto:
     """
     Helper function to recover a onnx model from its deserialized form
@@ -136,3 +122,17 @@ def mli_get_current_model(self) -> ColearnModel:
         Returns the current model
         """
         pass
+
+    @abc.abstractmethod
+    def mli_make_prediction(self, request: PredictionRequest) -> Prediction:
+        """
+        Make prediction using the current model.
+        Does not change the current weights of the model.
+
+        :param request: data to get the prediction for
+        :returns: the prediction
+        """
+        pass
+
+
+_DM_PREDICTION_SUFFIX = b">>>result<<<"

colearn/onnxutils.py

@@ -0,0 +1,44 @@
+# ------------------------------------------------------------------------------
+#
+#   Copyright 2021 Fetch.AI Limited
+#
+#   Licensed under the Creative Commons Attribution-NonCommercial International
+#   License, Version 4.0 (the "License"); you may not use this file except in
+#   compliance with the License. You may obtain a copy of the License at
+#
+#       http://creativecommons.org/licenses/by-nc/4.0/legalcode
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License 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.
+#
+# ------------------------------------------------------------------------------
+from typing import Any
+
+import onnxmltools
+import sklearn
+import tensorflow as tf
+import torch
+from tensorflow import keras
+
+model_classes_keras = (tf.keras.Model, keras.Model, tf.estimator.Estimator)
+model_classes_scipy = (torch.nn.Module)
+model_classes_sklearn = (sklearn.base.ClassifierMixin)
+
+
+def convert_model_to_onnx(model: Any):
+    """
+    Helper function to convert a ML model to onnx format
+    """
+    if isinstance(model, model_classes_keras):
+        return onnxmltools.convert_keras(model)
+    if isinstance(model, model_classes_sklearn):
+        return onnxmltools.convert_sklearn(model)
+    if 'xgboost' in model.__repr__():
+        return onnxmltools.convert_sklearn(model)
+    if isinstance(model, model_classes_scipy):
+        raise Exception("Pytorch models not yet supported to onnx")
+    else:
+        raise Exception("Attempt to convert unsupported model to onnx: {model}")

colearn_examples/ml_interface/keras_fraud.py

@@ -44,7 +44,7 @@
 input_classes = 431
 n_classes = 1
 loss = "binary_crossentropy"
-optimizer = tf.keras.optimizers.Adam
+optimizer = tf.keras.optimizers.legacy.Adam
 l_rate = 0.0001
 l_rate_decay = 1e-5
 batch_size = 10000

colearn_examples/ml_interface/mli_fraud.py

@@ -24,7 +24,8 @@
 import sklearn
 from sklearn.linear_model import SGDClassifier
 
-from colearn.ml_interface import MachineLearningInterface, Weights, ProposedWeights, ColearnModel, ModelFormat, convert_model_to_onnx
+from colearn.ml_interface import MachineLearningInterface, Prediction, PredictionRequest, Weights, ProposedWeights, ColearnModel, ModelFormat
+from colearn.onnxutils import convert_model_to_onnx
 from colearn.training import initial_result, collective_learning_round, set_equal_weights
 from colearn.utils.plot import ColearnPlot
 from colearn.utils.results import Results, print_results
@@ -130,6 +131,9 @@ def test(self, data, labels):
         except sklearn.exceptions.NotFittedError:
             return 0
 
+    def mli_make_prediction(self, request: PredictionRequest) -> Prediction:
+        raise NotImplementedError()
+
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser()

colearn_examples/ml_interface/mli_random_forest_iris.py

@@ -22,7 +22,8 @@
 from sklearn import datasets
 from sklearn.ensemble import RandomForestClassifier
 
-from colearn.ml_interface import MachineLearningInterface, Weights, ProposedWeights, ColearnModel, ModelFormat, convert_model_to_onnx
+from colearn.ml_interface import MachineLearningInterface, Prediction, PredictionRequest, Weights, ProposedWeights, ColearnModel, ModelFormat
+from colearn.onnxutils import convert_model_to_onnx
 from colearn.training import initial_result, collective_learning_round
 from colearn.utils.plot import ColearnPlot
 from colearn.utils.results import Results, print_results
@@ -114,6 +115,9 @@ def set_weights(self, weights: Weights):
     def test(self, data_array, labels_array):
         return self.model.score(data_array, labels_array)
 
+    def mli_make_prediction(self, request: PredictionRequest) -> Prediction:
+        raise NotImplementedError()
+
 
 train_fraction = 0.9
 vote_fraction = 0.05

colearn_examples/ml_interface/xgb_reg_boston.py

@@ -23,7 +23,8 @@
 import numpy as np
 import xgboost as xgb
 
-from colearn.ml_interface import MachineLearningInterface, Weights, ProposedWeights, ColearnModel, ModelFormat, convert_model_to_onnx
+from colearn.ml_interface import MachineLearningInterface, Prediction, PredictionRequest, Weights, ProposedWeights, ColearnModel, ModelFormat
+from colearn.onnxutils import convert_model_to_onnx
 from colearn.training import initial_result, collective_learning_round
 from colearn.utils.data import split_list_into_fractions
 from colearn.utils.plot import ColearnPlot
@@ -112,6 +113,9 @@ def mli_get_current_model(self) -> ColearnModel:
     def test(self, data_matrix):
         return mse(self.model.predict(data_matrix), data_matrix.get_label())
 
+    def mli_make_prediction(self, request: PredictionRequest) -> Prediction:
+        raise NotImplementedError()
+
 
 train_fraction = 0.9
 vote_fraction = 0.05

colearn_grpc/example_grpc_learner_client.py

@@ -24,7 +24,7 @@
 
 import colearn_grpc.proto.generated.interface_pb2 as ipb2
 import colearn_grpc.proto.generated.interface_pb2_grpc as ipb2_grpc
-from colearn.ml_interface import MachineLearningInterface, ProposedWeights, Weights, ColearnModel
+from colearn.ml_interface import MachineLearningInterface, Prediction, PredictionRequest, ProposedWeights, Weights, ColearnModel
 from colearn_grpc.logging import get_logger
 from colearn_grpc.utils import iterator_to_weights, weights_to_iterator
 
@@ -65,13 +65,15 @@ def start(self):
                     # Attempt to get the certificate from the server and use it to encrypt the
                     # connection. If the certificate cannot be found, try to create an unencrypted connection.
                     try:
-                        assert (':' in self.address), f"Poorly formatted address, needs :port - {self.address}"
+                        assert (
+                            ':' in self.address), f"Poorly formatted address, needs :port - {self.address}"
                         _logger.info(f"Connecting to server: {self.address}")
                         addr, port = self.address.split(':')
                         trusted_certs = ssl.get_server_certificate((addr, int(port)))
 
                         # create credentials
-                        credentials = grpc.ssl_channel_credentials(root_certificates=trusted_certs.encode())
+                        credentials = grpc.ssl_channel_credentials(
+                            root_certificates=trusted_certs.encode())
                     except ssl.SSLError as e:
                         _logger.warning(
                             f"Encountered ssl error when attempting to get certificate from learner server: {e}")
@@ -118,15 +120,21 @@ def get_supported_system(self):
         response = self.stub.QuerySupportedSystem(request)
         r = {
             "data_loaders": {},
+            "prediction_data_loaders": {},
             "model_architectures": {},
-            "compatibilities": {}
+            "data_compatibilities": {},
+            "pred_compatibilities": {},
         }
         for d in response.data_loaders:
             r["data_loaders"][d.name] = d.default_parameters
+        for p in response.prediction_data_loaders:
+            r["prediction_data_loaders"][p.name] = p.default_parameters
         for m in response.model_architectures:
             r["model_architectures"][m.name] = m.default_parameters
-        for c in response.compatibilities:
-            r["compatibilities"][c.model_architecture] = c.dataloaders
+        for dc in response.data_compatibilities:
+            r["data_compatibilities"][dc.model_architecture] = dc.dataloaders
+        for pc in response.pred_compatibilities:
+            r["pred_compatibilities"][pc.model_architecture] = pc.prediction_dataloaders
         return r
 
     def get_version(self):
@@ -137,18 +145,29 @@ def get_version(self):
         return response.version
 
     def setup_ml(self, dataset_loader_name, dataset_loader_parameters,
-                 model_arch_name, model_parameters):
-
-        _logger.info(f"Setting up ml: model_arch: {model_arch_name}, dataset_loader: {dataset_loader_name}")
+                 model_arch_name, model_parameters,
+                 prediction_dataset_loader_name=None,
+                 prediction_dataset_loader_parameters=None,
+                 ):
+        _logger.info(
+            f"Setting up ml: model_arch: {model_arch_name}, dataset_loader: {dataset_loader_name},"
+            f"prediction_dataset_loader: {prediction_dataset_loader_name}")
         _logger.debug(f"Model params: {model_parameters}")
         _logger.debug(f"Dataloader params: {dataset_loader_parameters}")
+        _logger.debug(
+            f"Prediction dataloader params: {prediction_dataset_loader_parameters}")
 
         request = ipb2.RequestMLSetup()
         request.dataset_loader_name = dataset_loader_name
         request.dataset_loader_parameters = dataset_loader_parameters
         request.model_arch_name = model_arch_name
         request.model_parameters = model_parameters
 
+        if request.prediction_dataset_loader_name:
+            request.prediction_dataset_loader_name = prediction_dataset_loader_name
+        if request.prediction_dataset_loader_parameters:
+            request.prediction_dataset_loader_parameters = prediction_dataset_loader_parameters
+
         _logger.info(f"Setting up ml with request: {request}")
 
         try:
@@ -173,7 +192,8 @@ def mli_propose_weights(self) -> Weights:
     def mli_test_weights(self, weights: Weights = None) -> ProposedWeights:
         try:
             if weights:
-                response = self.stub.TestWeights(weights_to_iterator(weights, encode=False))
+                response = self.stub.TestWeights(
+                    weights_to_iterator(weights, encode=False))
             else:
                 raise Exception("mli_test_weights(None) is not currently supported")
 
@@ -211,3 +231,19 @@ def mli_get_current_model(self) -> ColearnModel:
         response = self.stub.GetCurrentModel(request)
 
         return ColearnModel(model_format=response.model_format, model_file=response.model_file, model=response.model)
+
+    def mli_make_prediction(self, request: PredictionRequest) -> Prediction:
+        request_pb = ipb2.PredictionRequest()
+        request_pb.name = request.name
+        request_pb.input_data = request.input_data
+        if request.pred_data_loader_key:
+            request_pb.pred_data_loader_key = request.pred_data_loader_key
+
+        _logger.info(f"Requesting prediction {request.name}")
+
+        try:
+            response = self.stub.MakePrediction(request_pb)
+            return Prediction(name=response.name, prediction_data=response.prediction_data)
+        except grpc.RpcError as ex:
+            _logger.exception(f"Failed to make_prediction: {ex}")
+            raise ConnectionError(f"GRPC error: {ex}")