Hi, Iām @penhsuanwang
Researching on streaming data techniques and relative ML application.
Designed architecture based on Kafka & Spark platform.
Iām interested in state-of-the-art ML model for incremental learning.
License: MIT License
Hi, Iām @penhsuanwang
Researching on streaming data techniques and relative ML application.
Designed architecture based on Kafka & Spark platform.
Iām interested in state-of-the-art ML model for incremental learning.
Our application currently lacks the capability to list all models stored in the MLFlow model store directly through a web API. This feature is essential for enhancing our web UI's ability to display a comprehensive list of all model names along with their associated tags. Implementing this functionality will allow users to easily access and manage models through the web interface, promoting better interaction and accessibility.
Implement a new method in the MLFlowClientModelAgent
class that utilizes MlflowClient
from the MLflow tracking API to list all registered models. This method will support our new web API endpoint that provides a list of all models, which can then be used to populate the model management section of our web UI.
MLFlowClientModelAgent Class:
from mlflow.tracking import MlflowClient
class MLFlowClientModelAgent(MLFlowClient):
@classmethod
def list_all_models(cls):
"""
List all models registered in the MLflow model store.
This method will be used to provide data for a web API endpoint.
:return: A list of registered models with their names and tags.
"""
client = MlflowClient()
models = client.list_registered_models()
model_details = [{'name': model.name, 'tags': model.tags} for model in models]
return model_details
The enhancement allows users to manage the machine learning system more effectively by providing REST API endpoints to:
src/store/model_store.py
)src/store/data_processor_store.py
)src/store/trainer_store.py
)src/webapp/ml_training_serving_app.py
)src/webapp/ml_training_serving_app_router.py
)update_model(cls, model_id: str, new_model: object) -> bool
update_data_processor(cls, data_processor_id: str, new_data_processor: object) -> bool
update_trainer(cls, trainer_id: str, new_trainer: object) -> bool
MLTrainingServingApp
New Methods:
list_models(cls) -> list
list_trainers(cls) -> list
list_data_processors(cls) -> list
update_model(cls, model_id: str, model_params: dict) -> bool
update_trainer(cls, trainer_id: str, trainer_params: dict) -> bool
update_data_processor(cls, data_processor_id: str, data_processor_params: dict) -> bool
src/webapp/ml_training_serving_app_router.py
New Endpoints:
List Models:
@router.get("/ml_training_manager/list_models")
def list_models(ml_trainer_app: MLTrainingServingApp = Depends(get_app)):
# Implementation
List Trainers:
@router.get("/ml_training_manager/list_trainers")
def list_trainers(ml_trainer_app: MLTrainingServingApp = Depends(get_app)):
# Implementation
List Data Processors:
@router.get("/ml_training_manager/list_data_processors")
def list_data_processors(ml_trainer_app: MLTrainingServingApp = Depends(get_app)):
# Implementation
Update Model:
@router.put("/ml_training_manager/update_model/{model_id}")
def update_model(model_id: str, update_params: UpdateModelParams, ml_trainer_app: MLTrainingServingApp = Depends(get_app)):
# Implementation
Update Trainer:
@router.put("/ml_training_manager/update_trainer/{trainer_id}")
def update_trainer(trainer_id: str, update_params: UpdateTrainerParams, ml_trainer_app: MLTrainingServingApp = Depends(get_app)):
# Implementation
Update Data Processor:
@router.put("/ml_training_manager/update_data_processor/{data_processor_id}")
def update_data_processor(data_processor_id: str, update_params: UpdateDataProcessorParams, ml_trainer_app: MLTrainingServingApp = Depends(get_app)):
# Implementation
List Operations:
Update Operations:
Client -> Server: GET /ml_training_manager/list_models
Server -> ModelStore: list_models()
ModelStore -> Server: Return list of models
Server -> Client: Return list of models
Client -> Server: GET /ml_training_manager/list_trainers
Server -> TrainerStore: list_trainers()
TrainerStore -> Server: Return list of trainers
Server -> Client: Return list of trainers
Client -> Server: GET /ml_training_manager/list_data_processors
Server -> DataProcessorStore: list_data_processors()
DataProcessorStore -> Server: Return list of data processors
Server -> Client: Return list of data processors
Client -> Server: PUT /ml_training_manager/update_model/{model_id}
Server -> ModelStore: get_model(model_id)
ModelStore -> Server: Return model
Server -> Model: Update model parameters
Server -> ModelStore: update_model(model_id, updated_model)
ModelStore -> Server: Return update status
Server -> Client: Return update status
Client -> Server: PUT /ml_training_manager/update_trainer/{trainer_id}
Server -> TrainerStore: get_trainer(trainer_id)
TrainerStore -> Server: Return trainer
Server -> Trainer: Update trainer parameters
Server -> TrainerStore: update_trainer(trainer_id, updated_trainer)
TrainerStore -> Server: Return update status
Server -> Client: Return update status
Client -> Server: PUT /ml_training_manager/update_data_processor/{data_processor_id}
Server -> DataProcessorStore: get_data_processor(data_processor_id)
DataProcessorStore -> Server: Return data processor
Server -> DataProcessor: Update data processor parameters
Server -> DataProcessorStore: update_data_processor(data_processor_id, updated_data_processor)
DataProcessorStore -> Server: Return update status
Server -> Client: Return update status
We need to enhance our ML training application by adding support for initializing the data processor with a custom DataFrame directly passed by the user. This approach maintains backward compatibility with the existing data fetcher functionality while providing additional flexibility for users to directly supply their training data.
Update init_data_processor
Method:
init_data_processor
method in MLTrainingServingApp
to accept an optional DataFrame parameter.Minimal Impact on Existing Functionality:
Before Refactor:
class MLTrainingServingApp:
# Other methods...
@classmethod
def init_data_processor(cls, data_processor_type: str, **kwargs) -> bool:
"""
Design for an exposed REST api to let client init the data preprocessor
To initialize the data preprocessor
Initialize parameters from kwargs provided by client via REST api request body.
:param data_processor_type: The type of data processor to initialize
:param kwargs: Additional parameters for the data processor
:return: True if data processor is successfully initialized
"""
if cls._data_fetcher is None:
print("Data fetcher is not initialized")
return False
if not cls._raw_pandas_dataframe:
try:
cls._raw_pandas_dataframe = cls._data_fetcher.get_as_dataframe()
except ValueError as ve:
print("Failed to get data from data fetcher, try fetcher again")
cls._raw_pandas_dataframe = cls._data_fetcher.fetch_from_source()
cls._raw_pandas_dataframe = cls._data_fetcher.get_as_dataframe()
try:
cls._data_processor = DataProcessorFactory.create_data_processor(
data_processor_type,
input_data=cls._raw_pandas_dataframe,
**kwargs
)
cls._data_processor_store.add_data_processor(
data_processor_id="pytorch_lstm_aapl",
data_processor=cls._data_processor
)
except Exception as e:
print("Failed to init data processor")
return False
return True
After Refactor:
class MLTrainingServingApp:
# Other methods...
@classmethod
def init_data_processor(cls, data_processor_type: str, dataframe: pd.DataFrame = None, **kwargs) -> bool:
"""
Design for an exposed REST api to let client init the data preprocessor
To initialize the data preprocessor
Initialize parameters from kwargs provided by client via REST api request body.
If a DataFrame is provided, it will be used instead of fetching data from the data fetcher.
:param data_processor_type: The type of data processor to initialize
:param dataframe: Optional pandas DataFrame containing the data to process
:param kwargs: Additional parameters for the data processor
:return: True if data processor is successfully initialized
"""
if dataframe is not None:
# Use the provided DataFrame
cls._raw_pandas_dataframe = dataframe
else:
# Check if the data fetcher is initialized
if cls._data_fetcher is None:
print("Data fetcher is not initialized")
return False
if not cls._raw_pandas_dataframe:
try:
cls._raw_pandas_dataframe = cls._data_fetcher.get_as_dataframe()
except ValueError as ve:
print("Failed to get data from data fetcher, try fetcher again")
cls._raw_pandas_dataframe = cls._data_fetcher.fetch_from_source()
cls._raw_pandas_dataframe = cls._data_fetcher.get_as_dataframe()
try:
cls._data_processor = DataProcessorFactory.create_data_processor(
data_processor_type,
input_data=cls._raw_pandas_dataframe,
**kwargs
)
# Register the data processor to data processor manager
cls._data_processor_store.add_data_processor(
data_processor_id="pytorch_lstm_aapl",
data_processor=cls._data_processor
)
except Exception as e:
print("Failed to init data processor")
print(e)
return False
return True
The violation of the provided window size is occurring.
Need to check the minimum data size and the corresponding window size.
def _sliding_window_mask(self, input_array: np.ndarray) -> tuple[list[np.ndarray], list[np.ndarray]]:
"""
Sliding window mask on provided data, return training data and target data,
"""
x = []
y = []
for i in range(self._training_window_size, len(input_array) - self._target_window_size + 1):
x.append(input_array[i - self._training_window_size:i])
y.append(input_array[i:i + self._target_window_size, 0])
Provide a top down story to introduce modular design and functionality
The design of MLFlowClientModelLoader
, which is inherited from the base classes MLFlowClient
and MLFlowClientModelAgent
, could potentially cause a Method Resolution Order conflict when it needs to be integrated into mlflow_agent
in the future.
class MLFlowAgent(
MLFlowTracking,
MLFlowConfiguration,
MLFlowModelRegistry,
MLFlowClientModelLoader,
metaclass=SingletonMeta
):
need to fix in the future
We are updating the MLTrainingServingApp
to include functionality for managing trainers and models using separate singleton stores. This enhancement adheres to the Single Responsibility Principle and ensures a cleaner, more modular codebase. Additionally, we are adding new API endpoints to fetch existing trainers and models.
1. Introduced Separate Stores:
TrainerStore
: Manages trainer instances.ModelStore
: Manages model instances.2. Updated MLTrainingServingApp
:
TrainerStore
and ModelStore
.3. Added New API Endpoints:
Singleton Stores:
trainer_store.py
:
class TrainerStore:
# Singleton implementation
def add_trainer(cls, trainer_id: str, trainer: object) -> bool:
# Add trainer to store
def get_trainer(cls, trainer_id: str) -> object:
# Get trainer from store
model_store.py
:
class ModelStore:
# Singleton implementation
def add_model(cls, model_id: str, model: object) -> bool:
# Add model to store
def get_model(cls, model_id: str) -> object:
# Get model from store
MLTrainingServingApp Enhancements:
TrainerStore
and ModelStore
:
class MLTrainingServingApp:
def init_model(cls, model_type: str, model_id: str, **kwargs) -> bool:
# Initialize and store model
def init_trainer(cls, trainer_type: str, trainer_id: str, **kwargs) -> bool:
# Initialize and store trainer
def get_model(cls, model_id: str):
# Fetch model from store
def get_trainer(cls, trainer_id: str):
# Fetch trainer from store
New API Endpoints:
@router.get("/ml_training_manager/get_trainer/{trainer_id}")
def get_trainer(trainer_id: str):
# Fetch trainer by ID
@router.get("/ml_training_manager/get_model/{model_id}")
def get_model(model_id: str):
# Fetch model by ID
Create TrainerStore and ModelStore:
Update MLTrainingServingApp:
TrainerStore
and ModelStore
.Add New API Endpoints:
The web application implementation, ml_training_serving_app
, hardcodes the MLFlow Client settings in the function.
It has to be refactored to allow MLFlow Client settings to be passed from the client via an API.
@classmethod
def init_trainer(cls, trainer_type: str, **kwargs) -> bool:
"""
Design for an exposed REST api to let client init the trainer
To initialize the trainer
Initialize parameters from kwargs provided by client via REST api request body.
The trainer provide the function to add mlflow agent to log the training process and register the model
create mlflow agent and setting the tracking uri here.
:param trainer_type:
:param kwargs:
:return:
"""
criterion = None
if kwargs["loss_function"] == "mse":
criterion = torch.nn.MSELoss()
optimizer = None
if kwargs["optimizer"] == "adam":
optimizer = torch.optim.Adam(cls._model.parameters(), lr=float(kwargs["learning_rate"]))
mlflow_agent = MLFlowAgent()
> os.environ['MLFLOW_TRACKING_USERNAME'] = 'mlflow_pwang'
> os.environ['MLFLOW_TRACKING_PASSWORD'] = 'mlflow_pwang'
> mlflow_agent.set_tracking_uri("http://localhost:5011")
Rethinking the Modules of the ML Pipeline: Data Input/Output Types
To unify the data structure of data fetcher, data processor, ML trainer, etc., providing clear type hints.
To enhance our FastAPI backend for machine learning, aim to implement a centralized object, MLTrainingPipeline
, for managing the complete ML training process. This pipeline will be responsible for data fetching, processing, model training, and evaluation.
MLTrainingPipeline
and MLTrainingPipelineBuilder
classes.MLTrainingPipeline
class with methods for data preparation, training, and evaluation.MLTrainingPipelineBuilder
class for flexible pipeline construction./create_training_pipeline
to handle the construction and execution of the training pipeline.Currently use custom data processors todo data engineering and get a torch tensor which already to train. Fix it and wrapped this data by torch Dataset
TTo add a new module to store data processed in a singleton pattern.
Because the model inference stage still needs the data processor to invert the output value scale.
Since the data processor is currently implemented under MlTrainingServerApp
:
class MLTrainingServingApp:
"""
A singleton class to serve the model training process
provided the operation interface to client via REST api
1. operating the data fetcher by data_io_serving_app, another exported REST api to user to init the data fetcher.
provide the request body with information about fetch_data parameter
"""
_data_io_serving_app = src.webapp.data_io_serving_app.get_app()
# internal module tools for ml training job
_data_fetcher = None
> _data_processor = None
_trainer = None
This causes the data processor to be pruned once the training loop is finished.
We should build a data processor storage class in a singleton pattern so that the data processor can be stored and shared across different tasks.
Issue Description:
We need to implement a new API endpoint to allow external systems to trigger our ML training pipeline by directly sending training data. The endpoint must be secure, validate the incoming data, and initiate the training process asynchronously.
Tasks:
TrainingDataRequest
Pydantic model for data validation./ml_training/trigger
endpoint in FastAPI.BackgroundTasks
or Celery.Acceptance Criteria:
Notes:
Two new features to the MLflow-based machine learning system: Model Performance Monitoring and Model Retraining Automation. These features will enhance the system's capability to monitor model performance continuously and trigger automated retraining based on predefined criteria.
Features:
Model Performance Monitoring:
ModelPerformanceMonitor
class that logs performance metrics at regular intervals.# src/model_ops_manager/mlflow_agent/monitoring.py
# Import necessary modules and classes
class ModelPerformanceMonitor:
def __init__(self, model_name, model_version, interval=60):
# Initialize with model name, version, and monitoring interval
pass
def start_monitoring(self):
# Start the monitoring thread
pass
def monitor_performance(self):
# Continuously fetch and log metrics at defined intervals
pass
def fetch_metrics(self):
# Implement logic to fetch performance metrics
pass
def log_metrics(self, metrics):
# Log the fetched metrics
pass
Model Retraining Automation:
ModelRetrainer
class that automates retraining based on predefined criteria.# src/model_ops_manager/mlflow_agent/retrainer.py
# Import necessary modules and classes
class ModelRetrainer:
def __init__(self, retraining_script, criteria):
# Initialize with retraining script and criteria
pass
def check_and_retrain(self, model_name, model_version):
# Check if retraining criteria are met and retrain model if necessary
pass
def get_run_id(self, model_name, model_version):
# Get the run ID for a specific model version
pass
def should_retrain(self, metrics):
# Determine if retraining is needed based on metrics
pass
def retrain_model(self):
# Execute the retraining script
pass
Tasks:
ModelPerformanceMonitor
class in monitoring.py
.ModelRetrainer
class in retrainer.py
.A declarative, efficient, and flexible JavaScript library for building user interfaces.
š Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ššš
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ā¤ļø Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.