SIENTIAPDE-1171

Update dependencies and enhance ML model retraining functionality

- Updated sientia-dataops-library version in requirements.txt from 1.3.3 to 1.3.4.
- Incremented image tag in values.yaml from 0.2.4 to 0.2.5 and added a new environment variable MONGODB_TTL_INDEX_HOURS.
- Introduced new methods in MLFlowRepository for model retraining and production model updates, including error handling and logging.
- Added retrain_model and update_production_model activities in mlflow.py to support model management workflows.
- Modified MongoDB connection settings in connectors_config.py for improved security and configuration flexibility.

laborious/activities/mlflow.py

@@ -1,5 +1,3 @@
-import numpy as np
-from pandas import DataFrame
 from temporalio import activity, workflow
 
 
@@ -9,6 +7,9 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.temporal.utils.logger import Logger
     from laborious.utils.repository.model_repository import MLFlowRepository
     from typing import Any
+    import numpy as np
+    from pandas import DataFrame
+    import traceback
 
 
 class MLFlow(BaseActivity):
@@ -96,3 +97,107 @@ class MLFlow(BaseActivity):
         self.debug(response_data, metadata)
 
         return response_data
+
+    @activity.defn(name="retrain_model")
+    async def retrain_model(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Retrain the model.
+        Args:
+            - input_data (dict): The input data. Contains:
+                - model_name (str): The name of the model.
+                - data (dict[str, Any]): The data to retrain the model.
+        """
+        metadata = input_data['metadata']
+        data = DataFrame(input_data['data'])
+        model_name = input_data['model_name']
+
+        self.info(f'Retraining model {model_name}...', metadata)
+
+        timestamp = data['timestamp'].max()
+        self.debug(f'Timestamp: {timestamp}', metadata)
+
+        data.drop(columns=['model_id'], inplace=True, errors='ignore')
+        data.drop(columns=['created_at'], inplace=True, errors='ignore')
+
+        data = data.pivot(index='timestamp', columns='variable',
+                          values='value')
+        data.sort_index(inplace=True)
+        data.reset_index(inplace=True)
+
+        data = data.dropna()
+        data.columns.name = None
+
+        try:
+            retrain_output, experiment = self.model_monitoring_repository.retrain_model(
+                data=data,
+                model_name=model_name
+            )
+
+            return {
+                'status': retrain_output,
+                'timestamp': timestamp,
+                'experiment': experiment
+            }
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.send_notification(
+                metadata=metadata,
+                notification_id='RETRAIN_MODEL_ERROR',
+                message=f'Error retraining model {model_name}: {e}',
+                block='retrain_model',
+                attachment_content=trace
+            )
+            self.error(trace, metadata=metadata)
+            raise e
+
+    @activity.defn(name="update_production_model")
+    async def update_production_model(self, input_data: dict[str, Any]) -> dict[Any, Any]:
+        """
+        Update the production model.
+        Args:
+            - input_data (dict): The input data. Contains:
+                - model_name (str): The name of the model.
+                - experiment (str): The name of the experiment.
+                - model_id (str): The id of the model.
+                - timestamp (str): The timestamp of the model.
+                - status (str): The status of the model.
+        Returns:
+            dict[Any, Any]: The report of the model.
+        """
+        metadata = input_data['metadata']
+        model_name = input_data['model_name']
+        model_id = input_data['model_id']
+        experiment = input_data['experiment']
+        timestamp = input_data['timestamp']
+        status = input_data['status']
+
+        self.info(
+            f'Updating production model {model_name} from experiment {experiment}...', metadata)
+
+        try:
+            response = self.model_monitoring_repository.update_production_model(
+                experiment=experiment,
+                model_name=model_name
+            )
+
+            report = DataFrame([response])
+            report['model_id'] = model_id
+            report['model_name'] = model_name
+            report['timestamp'] = timestamp
+            report['status'] = status
+
+            self.info(
+                f'Production model {model_name} updated successfully', metadata)
+            return report.to_dict()
+
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.send_notification(
+                metadata=metadata,
+                notification_id='UPDATE_PRODUCTION_MODEL_ERROR',
+                message=f'Error updating production model {model_name}: {e}',
+                block='update_production_model',
+                attachment_content=trace
+            )
+            self.error(trace, metadata=metadata)
+            raise e

laborious/utils/connectors_config.py

@@ -47,12 +47,13 @@ def build_opc_config():
 
 
 def build_mongodb_config():
-    username = getenv('MONGODB_USERNAME', 'sientia')
-    password = getenv('MONGODB_PASSWORD', 'sientia')
-    uri = getenv('MONGODB_URL', 'localhost:27017')
+    username = getenv('MONGODB_USERNAME', 'root')
+    password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c')
+    uri = getenv('MONGODB_URL', 'localhost:27018')
 
     connection_string = f'mongodb://{username}:{password}@{uri}'
     return {
         'connection_string': connection_string,
-        'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia')
+        'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'),
+        'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600
     }

laborious/utils/repository/model_repository.py

@@ -13,6 +13,8 @@ By Monitoring we mean the evaluation of the performance of models, the generatio
 from datetime import datetime
 import traceback
 import pandas as pd
+import mlflow
+from os import makedirs, path, remove
 from sientia.ModelServing import ModelServing
 
 
@@ -85,3 +87,195 @@ class MLFlowRepository():
                     'traceback': traceback.format_exc()
                 }
             }
+
+    def get_experiment_by_run_id(self, run_id: str) -> dict:
+        # Get the run information using the run_id
+        run = mlflow.get_run(run_id)
+
+        # Extract the experiment ID from the run
+        experiment_id = run.info.experiment_id
+
+        # Get the experiment details using the experiment ID
+        experiment = mlflow.get_experiment(experiment_id)
+        experiment_name = experiment.name
+        return experiment_name
+
+    def get_next_run_name(self, model_name: str) -> str:
+        """
+        Function to get the next run number of a specific model
+
+        Parameters:
+        model_name (str): the name of the model
+
+        Returns:
+        str: the next run number
+        """
+
+        runs = mlflow.search_runs(
+            experiment_names=[model_name], order_by=["start_time desc"])
+        next_run_number = len(runs) + 1
+        return f"{model_name}-{next_run_number}"
+
+    def retrain_model(self, data: pd.DataFrame, model_name: str) -> tuple:
+        """
+        Retrain a model with new data.
+
+        Parameters:
+            data (pandas.DataFrame): The new data to use for retraining.
+            model_name (str): The name of the model to retrain.
+            metrics_list (list): The metrics to be used to compare the models.
+            compare_metrics (bool): If True, the retrain will only be considered if the new model is better than the current one.
+                                If False, the retrain will always be considered.
+            split_dataset (bool): If True, the data will be split into X and Y and into training and testing sets.
+                If False, the data will be used as a unique block for retraining.
+            update_report (bool): If True, a report will be created with the data of the retrained model.
+            update_transformation (bool): If True, the model will be updated in the MLflow tracking server.
+            update_prediction (bool): If True, the prediction model will be updated in the MLflow tracking server.
+            shuffle_data (bool): If True, the data will be shuffled before splitting.
+            model_type (str): The type of model to get metrics for. Ex: 'regression', 'classification'.
+
+
+        Returns:
+            mlflow.sklearn.Model: The retrained prediction model.
+            mlflow.sklearn.Model: The retrained data model.
+            mse (float): The mean squared error of the retrained model.
+            r2 (float): The R-squared score of the retrained model.
+        """
+
+        # load predictor model
+        predictor_uri = f"models:/{model_name}/production"
+        # load transform model
+        latest_production_id = self.model_serving.get_model_run_id(
+            model_name, stage="Production"
+        )
+        transform_uri = self.model_serving.get_model_uri(
+            latest_production_id, prediction=False
+        )
+        # load
+        data_model = mlflow.sklearn.load_model(transform_uri)
+        prediction_model = mlflow.sklearn.load_model(predictor_uri)
+        data_model = data_model.fit(data)
+        treated_data = data_model.predict(data)
+        # align target column with treated_data
+        target_name = data_model.target_variable
+        y = data[target_name]
+        treated_data = pd.merge(
+            treated_data, y, left_index=True, right_index=True)
+        prediction_model = prediction_model.fit(treated_data)
+        # Example usage
+        experiment = self.get_experiment_by_run_id(latest_production_id)
+        pred_model_atributes = vars(prediction_model)  # load class attributes
+        data_model_atributes = vars(data_model)  # load class attributes
+        mlflow.set_experiment(experiment)
+        experiment_description = "Retrain model {model_name} with new data"
+        current_run_name = self.get_next_run_name(experiment)
+        with mlflow.start_run(
+            run_name=current_run_name, description=experiment_description
+        ) as _run:
+            # update transfomation model
+            # fixed parameters
+            for name_atribute, val_atribute in pred_model_atributes.items():
+                if name_atribute != "model":
+                    mlflow.log_param(name_atribute, val_atribute)
+            # update prediction model
+            for name_atribute, val_atribute in data_model_atributes.items():
+                if name_atribute != "model":
+                    mlflow.log_param(name_atribute, val_atribute)
+            # dynamic parameters, including model itself
+            mlflow.sklearn.log_model(data_model, "data_model")
+
+            if not path.exists("temp"):
+                makedirs("temp")
+
+            file_path = f"temp/raw_data_{model_name}.csv"
+            data.to_csv(file_path, index=True)
+
+            # log the data raw
+            mlflow.log_artifact(file_path)
+
+            # dynamic parameters, including model itself
+            mlflow.sklearn.log_model(prediction_model, "prediction_model")
+            mlflow.log_param("retrain", True)
+
+            # clear temp file
+            if path.exists(file_path):
+                remove(file_path)
+
+            return "Model retrained successfully", experiment
+
+    def get_experiment(self, experiment_name: str) -> int:
+        experiment = mlflow.get_experiment_by_name(experiment_name)
+
+        if experiment is None:
+            raise ValueError(f'Experiment {experiment_name} not found')
+
+        return int(experiment.experiment_id)
+
+    def get_experiment_last_run(self, experiment_id: int) -> str:
+        runs = mlflow.search_runs(
+            experiment_ids=[experiment_id],
+            filter_string="",  # Sem filtro no MLflow ainda
+            output_format="pandas"
+        )
+
+        if not isinstance(runs, pd.DataFrame):
+            raise ValueError('Runs is not a pandas DataFrame')
+
+        # Filtrar apenas as runs onde params.retrain == True
+        filtered_runs = runs[runs["params.retrain"] == 'True']
+
+        # Converter a coluna 'end_time' para datetime
+        filtered_runs['end_time'] = pd.to_datetime(filtered_runs['end_time'])
+
+        # Ordenar o DataFrame de forma descendente pela coluna 'end_time'
+        filtered_runs = filtered_runs.sort_values(
+            by='end_time', ascending=False)
+
+        # Pegar a última run_id do DataFrame filtrado e ordenado
+        latest_run_id = filtered_runs.iloc[0]['run_id']
+
+        return latest_run_id
+
+    def update_production_model_by_run_id(self, run_id: str, model_name: str) -> dict:
+        # Registrar o modelo
+        # Aqui estamos assumindo que você já tem um modelo salvo, caso contrário você precisará treiná-lo e salvá-lo primeiro.
+        # Se o modelo já está registrado, você pode usar o método register_model() ou pyfunc.load_model() para isso.
+        mlflow.register_model(
+            f"runs:/{run_id}/prediction_model", model_name)
+
+        # Colocar a versão do modelo em produção
+        # Depois de registrar o modelo, precisamos pegar a versão mais recente do modelo e movê-lo para o estágio 'Production'
+        client = mlflow.tracking.MlflowClient()
+
+        # Obter a versão mais recente registrada do modelo
+        model_versions = client.get_registered_model(
+            model_name).latest_versions
+
+        if not isinstance(model_versions, list):
+            raise ValueError('Model versions is not a list')
+
+        max_version = max(model_versions, key=lambda x: int(x.version)).version
+
+        # Mover a versão mais recente do modelo para o estágio de 'Production'
+        client.transition_model_version_stage(
+            name=model_name,
+            version=max_version,
+            stage="Production",
+            archive_existing_versions=True
+        )
+
+        return {
+            'model_name': model_name,
+            'version': max_version,
+            'mlflow_run_id': run_id
+        }
+
+    def update_production_model(self, experiment: str, model_name: str) -> dict:
+
+        experiment_id = self.get_experiment(experiment)
+        run_id = self.get_experiment_last_run(experiment_id)
+        metadata = self.update_production_model_by_run_id(run_id, model_name)
+
+        metadata['mlflow_experiment_id'] = experiment_id
+
+        return metadata

laborious/worker/worker.py

@@ -6,6 +6,7 @@ with workflow.unsafe.imports_passed_through():
     import os
     import sys
     import asyncio
+    from laborious.workflows.minimal_retrain import MinimalRetrain
     from laborious.workflows.predictions_batch import PredictionsBatch
     from laborious.workflows.sub_workflows.prediction_process import PredictionProcess
     from laborious.workflows.sub_workflows.format_and_export_prediction import \
@@ -51,12 +52,28 @@ async def main():
 
     temporal_client = await client.Client.connect(
         target_host=host,
-        namespace=os.getenv('TEMPORAL_NAMESPACE', 'default')
+        namespace=os.getenv('TEMPORAL_NAMESPACE', 'laborious')
     )
 
     logger.info('Starting Workers...')
 
     workers = [
+        Worker(
+            temporal_client,
+            task_queue='minimal_retrain-queue',
+            workflows=[MinimalRetrain],
+            activities=[
+                activities.load_custom_query,
+                activities.retrain_model,
+                activities.update_production_model,
+                activities.export_data_to_postgres
+            ],
+            max_concurrent_workflow_tasks=100,
+            max_concurrent_activities=100,
+            max_concurrent_local_activities=100,
+            max_concurrent_workflow_task_polls=100,
+            max_cached_workflows=50,
+        ),
         Worker(
             temporal_client,
             task_queue='predictions_batch-queue',

laborious/workflows/minimal_retrain.py

@@ -0,0 +1,93 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from laborious.activities.activities import Activities
+    from typing import Any
+    from sientia_do.temporal.utils.policies import retry_policy
+    from datetime import timedelta
+
+
+@workflow.defn(name="minimal_retrain")
+class MinimalRetrain():
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]):
+        """
+        This workflow runs a minimal retrain of a model.
+
+        The workflow executes in four steps:
+        1. Loads the data from the database
+        2. Formats the data and perform the retrain
+        3. Updates the production model
+        4. Saves a model
+
+        Args:
+            - input_data (dict[str, Any]): The input data for the workflow.
+                - schedule_name (str): The name of the schedule.
+                - model_name (str): The name of the model.
+                - model_id (int): The id of the model.
+                - query (str): The SQL query to be executed to load data.
+                - schema (dict, optional): The schema to store the report.
+                - table_name (str, optional): The name of the table to store report.
+
+        Returns:
+            None
+
+        Raises:
+            Exception: If any of the required parameters are missing or if the workflow fails.
+        """
+
+        metadata = {
+            'metadata': {
+                'schedule_name': input_data['schedule_name'],
+                'model_name': input_data['model_name'],
+                'model_id': input_data['model_id'],
+                'workflow_name': 'minimal_retrain'
+            }
+        }
+
+        model_name = input_data['model_name']
+
+        data = await workflow.execute_local_activity_method(
+            Activities.load_custom_query,
+            {
+                **metadata,
+                'query': input_data['query'],
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        experiment_response = await workflow.execute_activity_method(
+            Activities.retrain_model,
+            {
+                **metadata,
+                'data': data,
+                'model_name': model_name
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        report = await workflow.execute_activity_method(
+            Activities.update_production_model,
+            {
+                **metadata,
+                'model_name': model_name,
+                'model_id': input_data['model_id'],
+                **experiment_response
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )
+
+        await workflow.execute_activity_method(
+            Activities.export_data_to_postgres,
+            {
+                **metadata,
+                'data': report,
+                'schema': input_data['schema'],
+                'table_name': input_data['table_name']
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )

requirements.txt

@@ -3,5 +3,5 @@ psycopg2-binary
 sqlalchemy
 asyncua
 redis
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.3
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.4
 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.1

values.yaml

@@ -11,7 +11,7 @@ image:
   # This sets the pull policy for images.
   pullPolicy: Always
   # Overrides the image tag whose default is the chart appVersion.
-  tag: "0.2.4"
+  tag: "0.2.5"
 
 # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
 imagePullSecrets:
@@ -178,6 +178,8 @@ env:
     value: "my-release-mongodb.mongodb.svc.cluster.local:27017"
   - name: MONGODB_DATABASE
     value: "sientia"
+  - name: MONGODB_TTL_INDEX_HOURS
+    value: "1"
 
 ssh:
   enabled: true