MLOps & 모델 라이프사이클 완전 정복: MLflow, DVC, LLMOps까지

# Kubeflow Pipeline 예시 - Level 1 CT 파이프라인
import kfp
from kfp import dsl

@dsl.component
def data_validation_op(data_path: str) -> bool:
    import great_expectations as ge
    ds = ge.read_csv(data_path)
    results = ds.expect_column_values_to_not_be_null("target")
    return results["success"]

@dsl.component
def train_model_op(data_path: str, model_output: str):
    import mlflow
    import sklearn
    # 모델 학습 로직
    pass

@dsl.pipeline(name="CT Pipeline")
def ct_pipeline(data_path: str):
    validation = data_validation_op(data_path=data_path)
    with dsl.Condition(validation.output == True):
        train_model_op(data_path=data_path, model_output="/models/")

소스 코드 변경 또는 데이터 트리거
        ↓
CI 파이프라인 (테스트, 빌드)
        ↓
CD 파이프라인 (파이프라인 배포)
        ↓
CT 파이프라인 (자동 재학습)
        ↓
모델 평가 → 통과/실패
        ↓
모델 레지스트리 등록
        ↓
Staging → Production 프로모션
        ↓
모니터링 & 알림

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score
import numpy as np

# MLflow Tracking 서버 설정
mlflow.set_tracking_uri("http://mlflow-server:5000")
mlflow.set_experiment("fraud-detection-v2")

with mlflow.start_run(run_name="rf-baseline") as run:
    # 하이퍼파라미터 로깅
    params = {
        "n_estimators": 100,
        "max_depth": 10,
        "min_samples_split": 5,
        "random_state": 42
    }
    mlflow.log_params(params)

    # 모델 학습
    model = RandomForestClassifier(**params)
    model.fit(X_train, y_train)

    # 메트릭 로깅
    y_pred = model.predict(X_test)
    metrics = {
        "accuracy": accuracy_score(y_test, y_pred),
        "f1_score": f1_score(y_test, y_pred, average="weighted"),
    }
    mlflow.log_metrics(metrics)

    # 모델 자동 로깅 (sklearn autolog 사용 시)
    # mlflow.sklearn.autolog()

    # 모델 저장
    mlflow.sklearn.log_model(
        sk_model=model,
        artifact_path="model",
        registered_model_name="fraud-detection",
        input_example=X_test[:5],
        signature=mlflow.models.infer_signature(X_train, y_pred)
    )

    # 커스텀 아티팩트
    import matplotlib.pyplot as plt
    from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay
    cm = confusion_matrix(y_test, y_pred)
    disp = ConfusionMatrixDisplay(confusion_matrix=cm)
    disp.plot()
    plt.savefig("confusion_matrix.png")
    mlflow.log_artifact("confusion_matrix.png")

    print(f"Run ID: {run.info.run_id}")
    print(f"Accuracy: {metrics['accuracy']:.4f}")

import mlflow

# 프레임워크 자동 감지 및 로깅
mlflow.autolog()

# PyTorch 전용 autolog
mlflow.pytorch.autolog(
    log_every_n_epoch=1,
    log_models=True,
    disable=False,
    exclusive=False,
    log_datasets=True
)

# XGBoost 전용 autolog
mlflow.xgboost.autolog(
    log_input_examples=True,
    log_model_signatures=True,
    log_models=True,
    log_datasets=True
)

# MLproject 파일
name: fraud-detection

conda_env: conda.yaml

entry_points:
  main:
    parameters:
      n_estimators: { type: int, default: 100 }
      max_depth: { type: int, default: 10 }
      data_path: { type: str, default: 'data/train.csv' }
    command: 'python train.py --n_estimators {n_estimators} --max_depth {max_depth} --data_path {data_path}'
  evaluate:
    parameters:
      model_uri: { type: str }
      test_data: { type: str }
    command: 'python evaluate.py --model_uri {model_uri} --test_data {test_data}'

import wandb
from wandb.integration.keras import WandbCallback

# W&B 초기화
run = wandb.init(
    project="image-classification",
    config={
        "learning_rate": 0.001,
        "epochs": 50,
        "batch_size": 32,
        "architecture": "ResNet50"
    }
)

# W&B Sweep으로 하이퍼파라미터 최적화
sweep_config = {
    "method": "bayes",
    "metric": {"name": "val_accuracy", "goal": "maximize"},
    "parameters": {
        "learning_rate": {"min": 1e-5, "max": 1e-2},
        "batch_size": {"values": [16, 32, 64]},
        "dropout": {"min": 0.1, "max": 0.5}
    }
}
sweep_id = wandb.sweep(sweep_config, project="image-classification")
wandb.agent(sweep_id, function=train_fn, count=50)

# DVC 초기화
git init
dvc init

# 원격 스토리지 설정 (S3)
dvc remote add -d myremote s3://my-bucket/dvc-store
dvc remote modify myremote region us-east-1

# 데이터 추가
dvc add data/train.csv
git add data/train.csv.dvc .gitignore
git commit -m "Add training data v1"
dvc push

# 다른 환경에서 데이터 가져오기
git pull
dvc pull

# dvc.yaml
stages:
  prepare:
    cmd: python src/prepare.py --input data/raw.csv --output data/processed/
    deps:
      - src/prepare.py
      - data/raw.csv
    outs:
      - data/processed/train.csv
      - data/processed/test.csv
    params:
      - prepare:
          - split_ratio
          - random_seed

  featurize:
    cmd: python src/featurize.py
    deps:
      - src/featurize.py
      - data/processed/train.csv
    outs:
      - data/features/train_features.pkl
    params:
      - featurize:
          - max_features
          - ngrams

  train:
    cmd: python src/train.py
    deps:
      - src/train.py
      - data/features/train_features.pkl
    outs:
      - models/model.pkl
    metrics:
      - reports/metrics.json:
          cache: false
    params:
      - train:
          - n_estimators
          - max_depth
          - random_seed

  evaluate:
    cmd: python src/evaluate.py
    deps:
      - src/evaluate.py
      - models/model.pkl
      - data/processed/test.csv
    metrics:
      - reports/eval_metrics.json:
          cache: false
    plots:
      - reports/plots/confusion_matrix.csv:
          cache: false

# params.yaml
prepare:
  split_ratio: 0.8
  random_seed: 42
featurize:
  max_features: 1000
  ngrams: 2
train:
  n_estimators: 100
  max_depth: 10
  random_seed: 42

# 파이프라인 실행
dvc repro

# 실험 브랜치 생성
dvc exp run --set-param train.n_estimators=200 --name exp-200-trees

# 실험 비교
dvc exp show

# 실험 결과 표 출력
dvc metrics show
dvc metrics diff

# feature_repo/feature_store.yaml
project: fraud_detection
registry: data/registry.db
provider: local
online_store:
  type: redis
  connection_string: "localhost:6379"
offline_store:
  type: bigquery
  dataset: feast_dev

# feature_repo/features.py
from datetime import timedelta
from feast import Entity, Feature, FeatureView, FileSource, ValueType
from feast.types import Float32, Int64, String

# 엔티티 정의
user = Entity(
    name="user_id",
    value_type=ValueType.INT64,
    description="사용자 ID"
)

# 데이터 소스 정의
user_stats_source = FileSource(
    path="data/user_stats.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created"
)

# 피처 뷰 정의
user_stats_fv = FeatureView(
    name="user_stats",
    entities=["user_id"],
    ttl=timedelta(days=7),
    features=[
        Feature(name="transaction_count_7d", dtype=Float32),
        Feature(name="avg_transaction_amount", dtype=Float32),
        Feature(name="days_since_last_login", dtype=Int64),
        Feature(name="account_age_days", dtype=Int64),
    ],
    online=True,
    source=user_stats_source,
    tags={"team": "fraud", "version": "v2"},
)

# 피처 스토어 사용 - 학습
from feast import FeatureStore
import pandas as pd

store = FeatureStore(repo_path="feature_repo/")

# 학습 데이터 조회 (오프라인)
entity_df = pd.DataFrame({
    "user_id": [1001, 1002, 1003],
    "event_timestamp": pd.to_datetime(["2026-03-01", "2026-03-01", "2026-03-01"])
})

training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_stats:transaction_count_7d",
        "user_stats:avg_transaction_amount",
        "user_stats:days_since_last_login",
    ]
).to_df()

# 온라인 서빙 - 실시간 피처 조회
feature_vector = store.get_online_features(
    features=[
        "user_stats:transaction_count_7d",
        "user_stats:avg_transaction_amount",
    ],
    entity_rows=[{"user_id": 1001}]
).to_dict()

from evidently import ColumnMapping
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

# 피처 드리프트 리포트 생성
report = Report(metrics=[DataDriftPreset()])
report.run(
    reference_data=reference_features,
    current_data=current_features,
    column_mapping=ColumnMapping(target="label")
)
report.save_html("feature_drift_report.html")

# 드리프트 결과 확인
results = report.as_dict()
drifted_features = [
    col for col, info in results["metrics"][0]["result"]["drift_by_columns"].items()
    if info["drift_detected"]
]
print(f"드리프트 감지된 피처: {drifted_features}")

import mlflow
from mlflow.tracking import MlflowClient

client = MlflowClient()

# 새 모델 등록
model_uri = f"runs:/{run_id}/model"
model_version = mlflow.register_model(
    model_uri=model_uri,
    name="fraud-detection"
)

# 모델 설명 추가
client.update_registered_model(
    name="fraud-detection",
    description="결제 사기 탐지 모델 - RandomForest 기반"
)

client.update_model_version(
    name="fraud-detection",
    version=model_version.version,
    description=f"Accuracy: 0.956, F1: 0.943 on test set"
)

# Staging으로 전환
client.transition_model_version_stage(
    name="fraud-detection",
    version=model_version.version,
    stage="Staging",
    archive_existing_versions=False
)

# Staging 모델 로드 및 검증
staging_model = mlflow.pyfunc.load_model(
    model_uri="models:/fraud-detection/Staging"
)
staging_preds = staging_model.predict(X_val)
staging_accuracy = accuracy_score(y_val, staging_preds)

# 검증 통과 시 Production 승격
if staging_accuracy > 0.95:
    client.transition_model_version_stage(
        name="fraud-detection",
        version=model_version.version,
        stage="Production",
        archive_existing_versions=True  # 이전 Production 버전 아카이브
    )
    print(f"모델 v{model_version.version} Production 승격 완료")

from huggingface_hub import HfApi, Repository
from transformers import AutoModelForSequenceClassification, AutoTokenizer

api = HfApi()

# 모델 업로드
api.upload_folder(
    folder_path="./fine-tuned-model",
    repo_id="myorg/sentiment-classifier-v2",
    repo_type="model",
)

# 모델 카드 업데이트
api.upload_file(
    path_or_fileobj="README.md",
    path_in_repo="README.md",
    repo_id="myorg/sentiment-classifier-v2",
)

# 특정 버전 태깅
api.create_tag(
    repo_id="myorg/sentiment-classifier-v2",
    tag="v2.1.0",
    tag_message="Improved accuracy on edge cases"
)

# .github/workflows/ml-cicd.yml
name: ML CI/CD Pipeline

on:
  push:
    branches: [main, develop]
    paths:
      - 'src/**'
      - 'params.yaml'
      - 'dvc.yaml'
  schedule:
    - cron: '0 2 * * 1' # 매주 월요일 오전 2시 자동 재학습

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Run unit tests
        run: pytest tests/ -v --cov=src
      - name: Data validation
        run: python src/validate_data.py

  train-and-evaluate:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Configure DVC remote
        run: |
          dvc remote modify myremote access_key_id ${{ secrets.AWS_ACCESS_KEY_ID }}
          dvc remote modify myremote secret_access_key ${{ secrets.AWS_SECRET_ACCESS_KEY }}
      - name: Pull data
        run: dvc pull
      - name: Run DVC pipeline
        run: dvc repro
      - name: Log metrics to MLflow
        run: python src/log_results.py
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}
      - name: Check model performance gate
        run: |
          python src/check_performance_gate.py \
            --min-accuracy 0.95 \
            --min-f1 0.93
      - name: Push results
        run: |
          dvc push
          git add reports/metrics.json dvc.lock
          git commit -m "chore: update metrics [skip ci]"
          git push

  deploy-staging:
    needs: train-and-evaluate
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
      - name: Promote model to Staging
        run: python src/promote_model.py --stage Staging
      - name: Run integration tests
        run: pytest tests/integration/ -v
      - name: Deploy to staging endpoint
        run: kubectl apply -f k8s/staging/

  deploy-production:
    needs: deploy-staging
    runs-on: ubuntu-latest
    environment: production
    steps:
      - name: Promote model to Production
        run: python src/promote_model.py --stage Production
      - name: Blue/Green deployment
        run: ./scripts/blue_green_deploy.sh
      - name: Smoke tests
        run: pytest tests/smoke/ -v

# src/check_retrain_trigger.py
import mlflow
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

def should_retrain(
    current_data,
    reference_data,
    performance_threshold=0.92,
    drift_threshold=0.3
) -> tuple[bool, str]:
    """재학습 필요 여부 판단"""

    # 1. 성능 기반 트리거
    current_metrics = get_current_metrics()
    if current_metrics["accuracy"] < performance_threshold:
        return True, f"성능 저하: accuracy={current_metrics['accuracy']:.3f}"

    # 2. 데이터 드리프트 트리거
    report = Report(metrics=[DataDriftPreset()])
    report.run(reference_data=reference_data, current_data=current_data)
    results = report.as_dict()
    drift_share = results["metrics"][0]["result"]["share_of_drifted_columns"]

    if drift_share > drift_threshold:
        return True, f"데이터 드리프트: {drift_share:.1%} 피처 드리프트 감지"

    return False, "재학습 불필요"

def get_current_metrics():
    client = mlflow.tracking.MlflowClient()
    prod_model = client.get_latest_versions("fraud-detection", stages=["Production"])[0]
    run = client.get_run(prod_model.run_id)
    return {
        "accuracy": float(run.data.metrics.get("accuracy", 0)),
        "f1_score": float(run.data.metrics.get("f1_score", 0))
    }

import pandas as pd
from evidently.report import Report
from evidently.test_suite import TestSuite
from evidently import ColumnMapping
from evidently.metric_preset import (
    DataDriftPreset,
    DataQualityPreset,
    TargetDriftPreset,
    ClassificationPreset
)
from evidently.tests import (
    TestNumberOfDriftedColumns,
    TestShareOfDriftedColumns,
    TestColumnDrift
)

# 컬럼 매핑 설정
column_mapping = ColumnMapping(
    target="fraud_label",
    prediction="fraud_score",
    numerical_features=["amount", "transaction_count_7d", "avg_amount"],
    categorical_features=["merchant_category", "payment_method"]
)

# 종합 드리프트 리포트
report = Report(metrics=[
    DataDriftPreset(),
    DataQualityPreset(),
    TargetDriftPreset(),
    ClassificationPreset()
])
report.run(
    reference_data=reference_df,
    current_data=production_df,
    column_mapping=column_mapping
)
report.save_html("monitoring/report.html")

# 알림 테스트 스위트
test_suite = TestSuite(tests=[
    TestNumberOfDriftedColumns(lt=3),
    TestShareOfDriftedColumns(lt=0.3),
    TestColumnDrift(column_name="amount"),
    TestColumnDrift(column_name="transaction_count_7d"),
])
test_suite.run(
    reference_data=reference_df,
    current_data=production_df
)

# 테스트 실패 시 알림
results = test_suite.as_dict()
failed_tests = [t for t in results["tests"] if t["status"] == "FAIL"]
if failed_tests:
    send_alert(f"모니터링 경보: {len(failed_tests)}개 테스트 실패")

# src/monitoring/metrics.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# 메트릭 정의
prediction_counter = Counter(
    "model_predictions_total",
    "총 예측 수",
    ["model_version", "result"]
)
prediction_latency = Histogram(
    "model_prediction_latency_seconds",
    "예측 지연 시간",
    buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0]
)
model_accuracy = Gauge(
    "model_accuracy_current",
    "현재 모델 정확도"
)
drift_score = Gauge(
    "feature_drift_score",
    "피처 드리프트 점수",
    ["feature_name"]
)

# 서빙 코드에서 메트릭 업데이트
def predict_with_monitoring(features, model_version="v2.1"):
    start_time = time.time()
    prediction = model.predict(features)
    latency = time.time() - start_time

    prediction_latency.observe(latency)
    prediction_counter.labels(
        model_version=model_version,
        result="fraud" if prediction[0] == 1 else "normal"
    ).inc()

    return prediction

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langsmith import Client
import os

# LangSmith 설정
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "production-chatbot"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"

# LangChain 체인 정의 (자동으로 LangSmith에 트레이스 기록)
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
prompt = ChatPromptTemplate.from_template(
    "당신은 친절한 고객 서비스 에이전트입니다.\n\n질문: {question}\n\n답변:"
)
chain = prompt | llm

# 실행 - 자동 트레이싱
response = chain.invoke({"question": "환불 정책이 어떻게 되나요?"})

# LangSmith 클라이언트로 평가
langsmith_client = Client()

# 데이터셋 생성
dataset = langsmith_client.create_dataset(
    dataset_name="customer-service-eval",
    description="고객 서비스 챗봇 평가 데이터셋"
)

# 평가 예제 추가
langsmith_client.create_examples(
    inputs=[{"question": "환불 정책을 알려주세요"}],
    outputs=[{"answer": "구매 후 30일 이내 환불 가능합니다."}],
    dataset_id=dataset.id
)

# 자동 평가 실행
from langsmith.evaluation import evaluate, LangChainStringEvaluator

evaluators = [
    LangChainStringEvaluator("cot_qa"),
    LangChainStringEvaluator("labeled_criteria", config={
        "criteria": "correctness"
    })
]
results = evaluate(
    lambda x: chain.invoke(x),
    data=dataset.name,
    evaluators=evaluators,
    experiment_prefix="gpt4o-baseline"
)

# prompt_registry.py
import mlflow
from dataclasses import dataclass
from typing import Optional

@dataclass
class PromptVersion:
    template: str
    version: str
    description: str
    metrics: Optional[dict] = None

class PromptRegistry:
    def __init__(self, mlflow_uri: str):
        mlflow.set_tracking_uri(mlflow_uri)
        self.experiment_name = "prompt-versions"
        mlflow.set_experiment(self.experiment_name)

    def register_prompt(self, prompt: PromptVersion) -> str:
        with mlflow.start_run(run_name=f"prompt-{prompt.version}") as run:
            mlflow.log_param("version", prompt.version)
            mlflow.log_param("description", prompt.description)
            mlflow.log_text(prompt.template, "prompt_template.txt")
            if prompt.metrics:
                mlflow.log_metrics(prompt.metrics)
            return run.info.run_id

    def get_prompt(self, version: str) -> str:
        client = mlflow.tracking.MlflowClient()
        runs = client.search_runs(
            experiment_ids=[mlflow.get_experiment_by_name(self.experiment_name).experiment_id],
            filter_string=f"params.version = '{version}'"
        )
        if not runs:
            raise ValueError(f"프롬프트 버전 {version}을 찾을 수 없습니다")
        artifact_uri = runs[0].info.artifact_uri
        return mlflow.artifacts.load_text(f"{artifact_uri}/prompt_template.txt")

# 사용 예시
registry = PromptRegistry("http://mlflow-server:5000")
registry.register_prompt(PromptVersion(
    template="당신은 {role}입니다. {context}\n\n질문: {question}\n답변:",
    version="v1.2.0",
    description="컨텍스트 포함 프롬프트 개선",
    metrics={"accuracy": 0.87, "hallucination_rate": 0.03}
))

# fine_tuning_pipeline.py
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    Trainer
)
from peft import LoraConfig, get_peft_model, TaskType
import mlflow

def fine_tune_with_lora(
    base_model: str,
    dataset_path: str,
    output_dir: str,
    lora_r: int = 16,
    lora_alpha: int = 32
):
    mlflow.set_experiment("llm-fine-tuning")

    with mlflow.start_run():
        # LoRA 설정
        lora_config = LoraConfig(
            task_type=TaskType.CAUSAL_LM,
            r=lora_r,
            lora_alpha=lora_alpha,
            target_modules=["q_proj", "v_proj"],
            lora_dropout=0.05,
            bias="none"
        )
        mlflow.log_params({
            "base_model": base_model,
            "lora_r": lora_r,
            "lora_alpha": lora_alpha
        })

        # 모델 준비
        model = AutoModelForCausalLM.from_pretrained(base_model)
        model = get_peft_model(model, lora_config)
        model.print_trainable_parameters()

        # 학습 설정
        training_args = TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=3,
            per_device_train_batch_size=4,
            gradient_accumulation_steps=4,
            learning_rate=2e-4,
            fp16=True,
            report_to="mlflow"
        )

        trainer = Trainer(
            model=model,
            args=training_args,
            train_dataset=train_dataset,
        )
        trainer.train()

        # 모델 저장 및 등록
        model.save_pretrained(output_dir)
        mlflow.transformers.log_model(
            transformers_model={"model": model, "tokenizer": tokenizer},
            artifact_path="fine-tuned-model",
            registered_model_name="customer-service-llm"
        )