Optimized serving endpoint for classical ML models

extras/iris_pipeline_project/.gitignore

@@ -0,0 +1 @@
+data/

extras/iris_pipeline_project/Dockerfile

@@ -0,0 +1,13 @@
+FROM openvino/model_server:latest
+
+USER root
+
+ENV LD_LIBRARY_PATH=/ovms/lib
+ENV PYTHONPATH=/ovms/lib/python
+
+RUN apt-get update && apt-get install -y python3-pip
+RUN pip3 install --break-system-packages pandas numpy scikit-learn joblib skl2onnx onnx onnxruntime
+
+
+WORKDIR /workspace
+ENTRYPOINT ["/ovms/bin/ovms", "--config_path", "/model_config.json", "--grpc_port", "9000"]

extras/iris_pipeline_project/client/client_inference.py

@@ -0,0 +1,25 @@
+import numpy as np
+import tritonclient.grpc as grpcclient
+import pandas as pd
+
+
+X_test = pd.read_csv("/home/harshitha/iris_pipeline_project/data/iris_test.csv")
+X_test = X_test.values.astype(np.float32)  
+
+input_name = "input"   
+output_name = "output_label" 
+
+inputs = []
+infer_input = grpcclient.InferInput(input_name, X_test.shape, "FP32")
+infer_input.set_data_from_numpy(X_test)
+inputs.append(infer_input)
+
+outputs = [grpcclient.InferRequestedOutput(output_name)]
+
+client = grpcclient.InferenceServerClient(url="localhost:9000")
+
+model_name = "iris_logreg"
+response = client.infer(model_name=model_name, inputs=inputs, outputs=outputs)
+
+predictions = response.as_numpy(output_name)
+print("Predictions:", predictions)

extras/iris_pipeline_project/client/client_train.py

@@ -0,0 +1,26 @@
+import numpy as np
+import pandas as pd
+import tritonclient.grpc as grpcclient
+
+data = pd.read_csv("/home/harshitha/iris_pipeline_project/data/iris_train.csv")
+df = pd.DataFrame(data)
+
+csv_str = df.to_csv(index=False)
+csv_bytes = np.frombuffer(csv_str.encode('utf-8'), dtype=np.uint8)
+
+input_name = "input"
+input = []
+infer_input = grpcclient.InferInput(input_name, csv_bytes.shape, "UINT8")
+infer_input.set_data_from_numpy(csv_bytes)
+input.append(infer_input)
+
+output_name = "output_label"
+output_label = [grpcclient.InferRequestedOutput(output_name)]
+
+client = grpcclient.InferenceServerClient(url="localhost:9000")
+
+model_name = "iris_pipeline"
+response = client.infer(model_name=model_name, inputs=input, outputs=output_label)
+
+result = response.as_numpy(output_name)
+print("Training result:", result)

extras/iris_pipeline_project/model_config.json

@@ -0,0 +1,9 @@
+{
+  "model_config_list": [],
+  "mediapipe_config_list": [
+    {
+      "name": "pipeline",
+      "graph_path": "/workspace/pipeline/graph.pbtxt"
+    }
+  ]
+}

extras/iris_pipeline_project/pipeline/graph.pbtxt

@@ -0,0 +1,15 @@
+input_stream: "OVMS_PY_TENSOR:pipeline_input"
+output_stream: "OVMS_PY_TENSOR:pipeline_output"
+
+node {
+  name: "python_node"
+  calculator: "PythonExecutorCalculator"
+  input_side_packet: "PYTHON_NODE_RESOURCES:py"
+  input_stream: "PIPELINE_INPUT:pipeline_input"
+  output_stream: "PIPELINE_OUTPUT:pipeline_output"
+  node_options: {
+    [type.googleapis.com/mediapipe.PythonExecutorCalculatorOptions]: {
+      handler_path: "/workspace/pipeline/ovmsmodel.py"
+    }
+  }
+}

extras/iris_pipeline_project/pipeline/ovmsmodel.py

@@ -0,0 +1,62 @@
+import numpy as np
+import pandas as pd
+import os
+import json
+from skl2onnx import convert_sklearn
+from skl2onnx.common.data_types import FloatTensorType
+from sklearn.linear_model import LogisticRegression
+import onnxruntime as ort
+
+MODEL_PATH = "/workspace/model/iris_logreg/1/model.onnx"
+LABEL_COLUMN = "species"
+
+class OvmsPythonModel:
+    def initialize(self, kwargs):
+        print("Training handler initialized.")
+
+    def execute(self, inputs, outputs, parameters, context):
+        # Expecting a dict: {"mode": "train" or "infer", "data": <CSV string>}
+        input_bytes = inputs["pipeline_input"]
+        try:
+            input_str = input_bytes.tobytes().decode('utf-8')
+            input_obj = json.loads(input_str)
+            mode = input_obj.get("mode")
+            csv_str = input_obj.get("data")
+        except Exception as e:
+            outputs["pipeline_output"] = np.array([f"ERROR: Invalid input format: {e}"], dtype=object)
+            return
+
+        try:
+            df = pd.read_csv(pd.compat.StringIO(csv_str))
+        except Exception as e:
+            outputs["pipeline_output"] = np.array([f"ERROR: Could not parse CSV: {e}"], dtype=object)
+            return
+
+        if mode == "train":
+            if LABEL_COLUMN not in df.columns:
+                outputs["pipeline_output"] = np.array([f"ERROR: Training data must include label column '{LABEL_COLUMN}'"], dtype=object)
+                return
+            X = df.drop(columns=[LABEL_COLUMN])
+            y = df[LABEL_COLUMN]
+            model = LogisticRegression(max_iter=200)
+            model.fit(X, y)
+            os.makedirs(os.path.dirname(MODEL_PATH), exist_ok=True)
+            initial_type = [('float_input', FloatTensorType([None, X.shape[1]]))]
+            onnx_model = convert_sklearn(model, initial_types=initial_type)
+            with open(MODEL_PATH, "wb") as f:
+                f.write(onnx_model.SerializeToString())
+            outputs["pipeline_output"] = np.array(["training complete"], dtype=object)
+        elif mode == "infer":
+            if not os.path.exists(MODEL_PATH):
+                outputs["pipeline_output"] = np.array(["ERROR: Model not trained yet"], dtype=object)
+                return
+            X = df.values.astype(np.float32)
+            sess = ort.InferenceSession(MODEL_PATH)
+            input_name = sess.get_inputs()[0].name
+            preds = sess.run(None, {input_name: X})[0]
+            outputs["pipeline_output"] = preds
+        else:
+            outputs["pipeline_output"] = np.array([f"ERROR: Unknown mode '{mode}'"], dtype=object)
+
+    def finalize(self):
+        print("Training handler finalized.")