From 89b9892a5b9d8a215c926593bdb7a0723f895573 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 23 Jul 2025 12:02:16 -0300 Subject: [PATCH] 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 | 109 +++++++++- laborious/utils/connectors_config.py | 9 +- .../utils/repository/model_repository.py | 194 ++++++++++++++++++ laborious/worker/worker.py | 19 +- laborious/workflows/minimal_retrain.py | 93 +++++++++ requirements.txt | 2 +- values.yaml | 4 +- 7 files changed, 421 insertions(+), 9 deletions(-) create mode 100644 laborious/workflows/minimal_retrain.py diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index bc0113e..63206a7 100644 --- a/laborious/activities/mlflow.py +++ b/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 diff --git a/laborious/utils/connectors_config.py b/laborious/utils/connectors_config.py index 5158f05..bcbb49b 100644 --- a/laborious/utils/connectors_config.py +++ b/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 } diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index b3473c6..86f1499 100644 --- a/laborious/utils/repository/model_repository.py +++ b/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 diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 005bb18..375c84f 100644 --- a/laborious/worker/worker.py +++ b/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', diff --git a/laborious/workflows/minimal_retrain.py b/laborious/workflows/minimal_retrain.py new file mode 100644 index 0000000..3d6f8d6 --- /dev/null +++ b/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) + ) diff --git a/requirements.txt b/requirements.txt index 8b1535e..fed750d 100644 --- a/requirements.txt +++ b/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 diff --git a/values.yaml b/values.yaml index 3a54067..bac9395 100644 --- a/values.yaml +++ b/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