MLFlow로 머신러닝 모델 실험 관리 및 Tag로 stage 표시

MLFlow란?

MLflow는 머신러닝 라이프사이클 관리 플랫폼으로, 실험 추적, 모델 개발, 배포, 그리고 운영까지 자동화하고 일관되게 관리할 수 있다. 실험 추적, 프로젝트 실행, 모델 저장 및 배포를 포함한 다양한 기능을 제공하여, 사용자가 재현 가능한 환경에서 모델을 관리할 수 있게 돕는다.

 

(좌) MLFlow 예시 화면 (우) MLFlow 파이프라인

 

(선택사항) 가상 환경 구축

venv 사용 방법

일단 파이썬 내장으로 실행할 수 있는 venv를 사용해서 가상 환경을 구축해보자. .venv로 앞에 .을 붙인 이유는 root 폴더에 생성되게 하기 위함이다. 실행은 두 번째 줄에 있는 source 명령어를 사용해서 가상 환경을 활성화시킬 수 있다.

python3.10 -m venv .venv
source .venv/bin/activate

활성화 되면 다음과 같이 terminal 상에서 (.venv)가 붙는다.

.venv 활성화 상태

가상환경 종료는 다음 명령어로 할 수 있다.

deactivate

venv 말고 conda를 이용해서도 가상 환경을 구축할 수 있다.

 

conda 사용 방법

conda create --name 'mlops-project' python=3.10
conda activate mlops-project

똑같이 가상환경 종료는 다음 명령어로 할 수 있다.

conda deactivate

 

MLFlow 설치

MLFlow 설치 명령어는 다음과 같다.

pip install mlflow

설치가 완료되었으면 다음 명령어로 ui를 확인해볼 수 있다.

mlflow ui

해당 명령어를 실행하면 링크가 뜨는데 터미널 상에서 cmd를 누르고 클릭하면 웹브라우저로 열린다.

(mlflow ui는 기본적으로 5000번 포트를 사용한다.)

mlflow ui 명령어 실행

(아래 화면 캡쳐는 5001 포트로 보이는데 내가 지금 5000포트 다른게 열려있어서 일단 5001로 포트를 지정해서 열었다.)

 

# mlflow ui 접속 포트 수정 명령어
mlflow ui --port 5001

mlflow ui 메인 화면

 

 

 

MLFlow 사용 코드

MLFlow import

이제 실제 모델링을 하면서 MLFlow가 어디서 어떻게 적용되는지 확인해보자.

import mlflow
import mlflow.sklearn

mlflow.__version__

이렇게 입력해서 mlflow가 잘 설치되었는지 확인해보자.
나는 2.16.2버전이 설치된걸 확인할 수 있다.

mlflow version 확인

sklearn 데이터셋을 이용하기 위해 모듈을 설치해도 되지만 사실 mlflow 의존성으로 scikit-learn, fastapi 등등이 이미 설치되어있어서 다음 명령어는 생략해도 된다.

!pip install scikit-learn

scikit-learn 설치 화면

 

데이터셋 준비

해당 내용은 데이터셋을 준비하고 모델링을 하는 과정이니 빠르게 넘어가도록 하겠다.

import pandas as pd
from sklearn.datasets import load_iris

iris = load_iris()

# df를 만들어거 모델 세팅하고 mlflow를 통해서 실험관리

df = pd.DataFrame(iris.data, columns=iris.feature_names)
df['Lable'] = iris.target
df

iris 데이터프레임

 

df.describe()

df.describe() 결과

 

from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score

X_train, X_test, y_train, y_test = train_test_split(iris.data, iris.target, test_size=0.2, random_state = 123)

model = LogisticRegression(max_iter=0)
model.fit(X_train, y_train) # 모의고사: X_train, y_train

pred = model.predict(X_test)
accuracy_score(y_test, pred)

이렇게 하면 정확도가 아마 0.5나 이렇게 낮게 나올것이다.

 

일단 여기서 max_iter를 0으로 고정한다고 했을때 우리가 취할수 있는 방법은 데이터셋을 스케일링하는 방법이 있을수 있다.

 

위에서 데이터 describe를 확인해봤을때 각 특성의 값 범위가 다르다.

• Sepal Length: 최소 4.3, 최대 7.9
• Sepal Width: 최소 2.0, 최대 4.4
• Petal Length: 최소 1.0, 최대 6.9
• Petal Width: 최소 0.1, 최대 2.5

Petal Length의 값은 1에서 6.9까지로 폭이 넓고, Petal Width는 0.1에서 2.5로 비교적 좁다. 이런 차이가 있으면, 특정 특성(범위가 큰 특성)이 모델에 더 큰 영향을 줄 수 있다. 예를 들어, Petal Length는 다른 특성들보다 큰 값 범위를 가지므로, 모델이 그 특성에 더 큰 가중치를 줄 수 있다.

 

스케일링을 통해 모든 특성의 값 범위를 동일한 범위(예: 0에서 1 사이)로 맞추면, 모델이 한 특성에 치우치지 않고 모든 특성을 균등하게 학습할 수 있다.

from sklearn.preprocessing import StandardScaler

scaler = StandardScaler()

X_scaled = scaler.fit_transform(iris.data)

X_train, X_test, y_train, y_test = train_test_split(X_scaled, iris.target, test_size=0.2, random_state=123)

model = LogisticRegression(max_iter=0)
model.fit(X_train, y_train)

pred = model.predict(X_test)
accuracy_score(y_test, pred)

이렇게 하니까 정확도가 0.9333333333333333 까지 올라갔다.

위에서 스케일링을 적용했을때 실제로 값이 어떻게 스케일링 되었는지는 다음 코드로 확인할 수 있다.

df = pd.DataFrame(X_scaled, columns=iris.feature_names)
df.describe()

스케일링된 데이터 확인

 

 

모델 실험 파라미터 및 모델 성능 기록

이제 MLFlow를 시용해서 모델 실험 파라미터와 모델의 성능을 기록해보자

MLflow가 실험 데이터를 추적하고 저장할 서버의 URI(주소)를 설정해주자.

지금 실습같은 경우에는 위에서 나온 uri를 적어주면 되는데 실무에서는 dev.회사명.com 같이 환경변수로 설정한 값을 넣어줄 수도 있다.

mlflow.set_tracking_uri('http://127.0.0.1:5000')
print(mlflow.get_tracking_uri())
>>> http://127.0.0.1:5000

이제 실험을 생성해보자.

exp = mlflow.set_experiment(experiment_name='iris_classifiacion')

display(exp.name)
display(exp.experiment_id)
display(exp.artifact_location)
display(exp.creation_time)   # 1726803122469

이렇게 실험의 이름을 작성해주고 실행하면 실험이 생성되고 해당 실험의 데이터들을 확인할 수 있다.

새로 생성된 Experiments

위에서 출력된 creation_time이 이상한 값으로 나오는데 그 값은 다음과 같이 확인해볼 수 있다.

from datetime import datetime

datetime.fromtimestamp(exp.creation_time/1000)
>>> datetime.datetime(2024, 9, 20, 12, 32, 2, 469000)

이제 위에서 만든 실험 실행(run)을 시작해보자.

mlflow.autolog()    # mlflow에서 자동으로 실험으 ㅣ메트릭과 파라미터를 기록해준다.

mlflow.start_run()  # start_run 이후 모델을 불러오고, 실험하는 코드들을 작성하면 된다.

model = LogisticRegression(max_iter=0)  # default=100
model.fit(X_train, y_train)
pred = model.predict(X_test)
accuracy = accuracy_score(y_test, pred)
print(f'정확도: {accuracy*100}')

mlflow.end_run()    # 실험 종료

이렇게 start_run()과 end_run() 사이에 모델링하는 코드를 작성하면 된다.

실행하면 다음과 같이 나온다.

실험 종료된 출력창

물론 이렇게 매번 start, end를 작성할수도 있지만 python의 with 구문을 사용하면 더 손쉽게 실험을 관리할 수 있다.

mlflow.autolog()    # 알아서 필요한 데이터 기록해달라

with mlflow.start_run():
    model = LogisticRegression(max_iter=0)  # default=100
    model.fit(X_train, y_train)
    pred = model.predict(X_test)
    accuracy = accuracy_score(y_test, pred)
    print(f'정확도: {accuracy*100}')

실험 상세 목록

 

해당 실험을 클릭하면 다음과 같은 화면을 볼 수 있다.

실행 상세 화면

왼쪽에 보이는 Parameters는 모델에 대한 파라미터들이다.

여러개의 모델을 실험

지금까지는 하나의 모델을 학습해봤다면 여러개의 모델을 이용해서 실험하는것도 기록이 가능하다.

from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.svm import SVC

# 알고리즘별 성능 테스트

models = {
    "LogisticRegression": LogisticRegression(
        max_iter=0,
        C = 1.0,
        solver='lbfgs',
        random_state=123
    ), 
    "RandomForestClassifier": RandomForestClassifier(
        n_estimators=100,
        random_state=123
    ),
    "SVC": SVC(
        kernel='linear',
        random_state=123
    )
}


for model_name, model in models.items():
    # print(model_name, model)
    model.fit(X_train, y_train)
    pred = model.predict(X_test)
    accuracy = accuracy_score(y_test, pred)

    mlflow.log_param('log_param_key', 'log_param_value')
    mlflow.log_metric('log_metric_key', 0.01)

    print(f'모델명: {model_name}, 정확도: {accuracy*100}')

이렇게 하면 여러개의 모델도 트랙킹 가능하다.

여러 모델들의 정확도 출력

중간에 사용된 log_param(), log_metric()은 mlflow ui 화면에서 확인할 수 있다.

• log_param(): 실험에서 사용된 파라미터를 기록한다. 모델 학습에 사용된 하이퍼파라미터나 설정 값을 기록하여, 나중에 어떤 설정이 사용되었는지 추적할 수 있다.
• log_metric(): 실험에서 얻은 메트릭(성능 지표)을 기록한다. 모델의 성능을 나타내는 지표(예: 정확도, 손실, F1 스코어 등)를 기록하여, 실험 간 비교가 가능하게 한다.

(지금은 예시로 임시값을 넣었지만 실제 실험 파라미터와 메트릭을 기록할 때는 해당 모델에 맞는 하이퍼파라미터나 정확한 메트릭을 기록하는 것이 좋다.)

Columns > Metrics & Parameters에서 확인 가능

지금까지 테스트를 해본 결과 Run Name이 무작위로 설정된걸 확인할 수 있다. 이 Run Name 또한 따로 커스텀해서 설정해줄 수 있다.

Run Name 설정

# run name 설정해주기
from datetime import datetime
today = datetime.today().strftime('%Y-%M-%D')

mlflow.autolog()    

for model_name, model in models.items():

    with mlflow.start_run(run_name=f'{today} - {model_name}', nested=True):
        model.fit(X_train, y_train)
        pred = model.predict(X_test)
        accuracy = accuracy_score(y_test, pred)

        mlflow.log_param('log_param_key', 'log_param_value')
        mlflow.log_metric('log_metric_key', 0.01)

        print(f'모델명: {model_name}, 정확도: {accuracy*100}')

nested=True 옵션을 설정해줬기 때문에 ui 화면에서 + 버튼을 클릭해야 제대로 보인다.

커스텀한 Run Name

 

MLFlow 모델 관리

MLFlow 모델을 관리한다는 뜻은 Stage를 관리하겠다라는 뜻이다.

기존에는 mlflow에서 stage 옵션으로 관리했었는데 이 기능이 deprecated되었다. 원래는 staging, production, archive 세 단계로 관리를 했었는데 이제는 회사 자체적으로 stage 단계를 설정하고 거기에 맞게 자율적으로 설정해라! 라는 의미로 기능을 없앤것같다.

그래서 우리는 기존에 사용하던 staging, production, archive를 Tag로써 관리해주겠다.

모델 등록

일단 모델을 등록해보자.

from mlflow.tracking import MlflowClient

client = MlflowClient()

# 모델 등록
def model_register(model_name, run_id):
    model_uri = f'runs:/{run_id}/model'
    model_version = mlflow.register_model(model_uri, model_name)
    return model_version    # workflow management

# 모델 등록
# http://127.0.0.1:5000/#/experiments/841097937802182006/runs/6b36de56cd8a441fbc9951f372971230

# 실험 id: 841097937802182006
# 실행 id: 6b36de56cd8a441fbc9951f372971230
run_id = "6b36de56cd8a441fbc9951f372971230"
model_name='RandomForestClassifier' # tags에서 가져옴
model_register(model_name, run_id)

모델 등록 완료

 

위 코드에서 가져온 uri는 실험 상세 페이지의 uri와 동일하다.

experiments/ 뒤에 나오는게 실험 ID이고, runs/ 뒤에 나오는게 실행 ID이다. (사실 이 실행 ID가 모델ID와 유사하다. 확정지어 모델 식별자라고 말하긴 조금 뭐하지만 모델을 식별하는 키이기도 한것같다. 모델별로 run이 생성되니까,,?)

이렇게 하면 이제 모델이 등록된다!

(캡쳐에서는 Version 6으로 보이는데 이는 내가 이미 테스트를 전부 다 하고 캡쳐한거라 그렇다. Version 1로 보이면 정상이다.)

Models > 등록한 모델

사실 모델을 등록하는것에서 끝나지 않고 대부분 모델을 등록하면 바로 staging으로 승격시킨다. 그래서 위에서 만든 '모델 등록' 함수를 'staging 승격' 함수에 포함시키자.

모델 등록 및 staging으로 승격

def promote_to_staging(model_name, run_id):
    model_version = model_register(model_name, run_id)
    client.set_model_version_tag(
        name=model_name,
        version=model_version.version,
        key='stage',
        value='staging'
    )

    print(f'Model: {model_name}, Version: {model_version.version} is promoted to Staging..')

promote_to_staging(model_name, run_id)

 

그럼 ui 에서 Tags에 stage: staging으로 표시가 된다.
(캡쳐본에서 Version 은 신경쓰지 말자,, 순서가 꼬여서 일단 임시로 진행한거다..)

stage: staging 표시

이제 staging과 동일하게 production 으로 승격시키는 함수도 만들어주자

production으로 승격

def promote_to_production(model_name, version):
    client.set_model_version_tag(
        name=model_name,
        version=version,
        key='stage',
        value='production'
    )
    
	print(f'Model: {model_name}, Version: {version} is promoted to Productions..')

promote_to_production(model_name, '6')

이렇게 production 함수에서는 직접 version을 입력하는건, staging 단계에서 production으로 승격시키면 실제로 바로 유저들이 이용하는 모델이 되기 때문에 자동으로 승격시키기보다는 적절한 모델인지 확인하고 여러 사람들의 리뷰를 받고 진행하는거기때문에 이부분은 일부러 수동으로 진행하도록 했다.

 

이렇게 했을때 새로운 버전이 production이 되면 그 이전 버전의 태그는 Archieved로 "수동으로" 바꿔줘야 한다.

archived 관리

def archive_model(model_name, version):
    client.set_model_version_tag(
        name=model_name,
        version=version,
        key='stage',
        value='archive'
    )

    print(f'Model: {model_name}, Version: {version} is moved to Archive..')

archive_model(model_name, '5')

그럼 대략 이런 모습을 띌 수 있다.

다양한 stage

(이렇게 진행한건 실제로 배포가 이루어지고 이런게 아니다. 화면상에서 직관적으로 이해하기 위해 시각적 표시를 한거지 실제 배포는 또 따로 이루어져야한다.)

 

Model Serving 모델 서빙

모델 서빙은 학습된 모델을 Flask나 FastAPI 같은 웹 프레임워크를 이용해 REST API 형태로 만들어서 클라이언트가 요청을 보내면 예측 값을 반환하는 구조다. MLFlow에서는 모델 서빙 기능을 제공하긴해서 이 과정을 간편하게 할 수 있다.(mlflow.models.serve)

 

우리는 API로 서빙하는 방법을 실습해보자.

 

모델 버전 관리와 Inference

MLFlow의 pyfunc 모듈을 이용해 모델을 로드하고, 버전까지 지정할 수 있다. 이때 모델 서빙에서 **inference**란, 입력 데이터를 API로 받아서 그에 대한 예측 값을 반환하는 과정을 의미한다.

model_name='RandomForestClassifier'
model_version='6'
model_uri=f'models:/{model_name}/{model_version}'    # model_uri는 model의 고유 ID라고 이해하면 된다.

loaded_model = mlflow.pyfunc.load_model(model_uri)  
loaded_model

# 예측 수행
test_input = X_test[:10]
loaded_model.predict(test_input)    # 주로 streamlit을 통해 PM, PO, 기획자에게 시각적으로 보여줄 수 있다.
>>> array([1, 2, 2, 1, 0, 1, 1, 0, 0, 1])

 

모델을 API로 서빙하기

이제 로드된 모델을 Flask나 FastAPI처럼 API로 요청했을 때, 예측 결과를 반환하는 방식으로 서빙할 수 있다.

# mlflow 설치할때 Flask가 함께 설치되었다. 즉, 서버를 하나 더 운영해야한다.
# http://127.0.0.1:5000/#/experiments/841097937802182006/runs/6b36de56cd8a441fbc9951f372971230/artifacts/model

# 841097937802182006 : 실험 아이디
# 6b36de56cd8a441fbc9951f372971230: 실행 아이디

# API 서버 실행 명령어
mlflow models serve -m ./mlartifacts/841097937802182006/6b36de56cd8a441fbc9951f372971230/artifacts/model -p [비어있는 port] --no-conda

 

--no-conda 옵션을 붙이면 conda 환경을 사용하지 않고, 서버가 더욱 가볍게 동작하도록 기본 환경에서 모델을 서빙하는 방식이다.

 

위 코드에서 완성된 uri를 알수 있는 방법은

1) MLFlow UI에서 서빙할 모델의 버전을 클릭

등록된 모델에서 버전 선택

2) Source Run을 클릭

Source Run 클릭

3) 페이지에서 해당 모델의 uri를 복사

uri 복사

URI를 이렇게 찾으면, models:/{model_name}/{model_version} 형태로 알맞게 모델을 로드할 수 있다.

위 숫자는 이렇게 지금 실행중인 ipynb 파일 트리를 보면 이해하기 쉽다.

vscode 파일 트리

이제 모델을 서빙해보자.

아래 명령어를 실행할때 아까 켜두었던 mlflow 터미널에서 진행하지 말고 새로운 터미널창에서 해당 명령어를 실행해야한다!!

mlflow models serve -m ./mlartifacts/841097937802182006/6b36de56cd8a441fbc9951f372971230/artifacts/model -p 5001 --no-conda

명령어 실행 모습

 

위 명령어를 사용하면 지정된 포트에서 Flask 서버가 실행되고, 해당 모델이 API로 서빙된다. 이제 클라이언트는 이 서버에 요청을 보내어 예측 결과를 받을 수 있다.

테스트 방법1. request 모듈 사용

실제로 모뎅이 잘 배포되었는지 request 모듈을 사용해서 확인해 볼 수 있다.

X_test_df = pd.DataFrame(X_test, columns=iris.feature_names)

X_test_df[:1].to_dict(orient='split')

{'index': [0],
 'columns': ['sepal length (cm)',
  'sepal width (cm)',
  'petal length (cm)',
  'petal width (cm)'],
 'data': [[0.5533332750260058,
   -1.282963310614858,
   0.6490834153533465,
   0.3957741007661703]]}

import requests

url = 'http://127.0.0.1:5001/invocations'
headers = {'Content-Type':'application/json'}
data = {
    'dataframe_split': X_test_df[:5].to_dict(orient='split')

res = requests.post(url=url, headers=headers, data=json.dumps(data)) # payload에 data가 포함된 형태. get은 data가 없고, url 기반으로 소통
>>> <Response [200]>

이렇게 일단 서버가 잘 켜져있는 모습을 확인할 수 있다.

res.json()
>>> {'predictions': [1, 2, 2, 1, 0]}

그리고 이렇게 실제 예측값도 받아올 수 있다.

테스트 방법 2. Thunder Client 익스텐션 사용

vscode에서 postman과 비슷한 Thunder Client 익스텐션을 사용해서 테스트해볼수도 있다.

Thunder Client 익스텐션

이렇게 익스텐션을 설치하고 uri에 실행한 서버의 주소를 작성해준다.

전송 방법은 POST로 선택해주고, Header에 위에서 설정해준것처럼 content-type을 작성해준다.

Thunder Client 익스텐션 사용 1

그리고 Body 탭으로 이동해서 위에서 담은 데이터를 동일하게 작성해준다.

Thunder Client 익스텐션 사용 2

그리고 Send 버튼을 누르면 제대로된 응답값을 받을 수 있다.

 

이렇게 간단하게 MLFlow 파이프라인을 살펴봤다. 더 많은 기능들이 있지만 여기서 마무리하려고 한다.