> ## Documentation Index > Fetch the complete documentation index at: https://docs.sifi.life/llms.txt > Use this file to discover all available pages before exploring further. # 커스텀 모델 기초: 시그니처와 PyFunc > 이 문서는 **[머신러닝](/training/09-machine-learning/overview)** 섹션의 일부입니다. ## 왜 커스텀 모델 배포가 필요한가? Databricks Model Serving은 등록된 MLflow 모델을 REST API 엔드포인트로 배포하는 관리형 서비스입니다. **Foundation Model API (FMAPI)** 가 제공하는 사전 학습 모델(llama, mixtral 등)로 충분하지 않은 경우, 커스텀 모델 배포가 필요합니다. ### Foundation Model API vs. 커스텀 모델 배포 | 구분 | Foundation Model API | 커스텀 모델 배포 | | ----------- | ----------------------- | -------------------- | | **모델 제어** | Databricks가 제공하는 모델만 사용 | 직접 학습/파인튜닝한 모델 사용 가능 | | **전처리/후처리** | 불가 (입출력 형식 고정) | 완전한 자유도 | | **비용** | 토큰당 종량제 (Pay-per-token) | 인스턴스 시간당 과금 | | **운영 복잡도** | 낮음 (인프라 자동 관리) | 높음 (의존성, 스케일링 직접 설정) | | **레이턴시** | 공유 인프라 (변동 가능) | 전용 인스턴스 (예측 가능) | | **규정 준수** | 데이터가 외부 모델 통과 | 워크스페이스 내부에서 처리 | ### 커스텀 배포가 필요한 경우 | 사용 사례 | 설명 | | --------------- | --------------------------------------- | | **전후처리 파이프라인** | 입력 데이터 정규화, 피처 엔지니어링, 출력 후처리가 필요한 경우 | | **파인튜닝 모델** | 도메인 특화 데이터로 Fine-tuning한 모델을 배포해야 하는 경우 | | **자체 학습 모델** | 사내 데이터로 처음부터 학습한 ML/DL 모델 | | **앙상블 모델** | 여러 모델의 예측을 결합해야 하는 경우 | | **외부 라이브러리 의존** | 표준 MLflow flavor에 없는 프레임워크를 사용하는 경우 | | **비즈니스 로직 포함** | 예측값에 규칙 기반 로직을 적용해야 하는 경우 | | **GPU 추론** | 대규모 딥러닝 모델을 GPU로 서빙해야 하는 경우 | | **데이터 거버넌스** | 입력 데이터가 외부 서비스로 나가면 안 되는 경우 | > **모델 서빙 엔드포인트 (Model Serving Endpoint)** 란 학습된 ML 모델을 REST API로 노출하여, 애플리케이션에서 실시간으로 예측을 요청할 수 있게 해주는 서비스입니다. Databricks에서는 인프라 관리, 오토스케일링, 모니터링을 자동으로 처리합니다. *** ## 배포 가능한 모델 유형 | 배포 유형 | 설명 | 포함 모델 | | -------------- | --------------------------------------------------- | --------------------------------- | | **표준 Flavor** | 기본 제공 MLflow Flavor를 사용합니다 | sklearn, xgboost, lightgbm, spark | | **커스텀 PyFunc** | `mlflow.pyfunc.PythonModel`을 상속하여 자유로운 추론 로직을 구현합니다 | 전처리/후처리, 멀티모델 앙상블 등 | | **외부 모델** | 외부 LLM 제공사를 프록시합니다 | OpenAI, Anthropic, Cohere | | 모델 유형 | 특징 | 적합한 경우 | | --------------------------- | ------------------------------------------------- | ------------------------------ | | **표준 MLflow Flavor** | `mlflow.sklearn`, `mlflow.xgboost` 등으로 저장한 모델 | 단순 predict 호출로 충분한 경우 | | **커스텀 PyFunc** | `mlflow.pyfunc.PythonModel`을 상속하여 자유로운 추론 로직 구현 | 전후처리, 앙상블, 비즈니스 로직이 필요한 경우 | | **외부 모델 (External Models)** | OpenAI, Anthropic 등 외부 API를 Databricks 엔드포인트로 프록시 | 외부 LLM API에 거버넌스/레이트 리밋을 적용할 때 | *** ## 모델 시그니처 정의 (Model Signature) **모델 시그니처 (Model Signature)** 는 모델의 입력/출력 스키마를 명시적으로 정의합니다. 시그니처가 있으면 서빙 시 입력 데이터 유효성 검사가 자동으로 수행되어 오류를 조기에 감지할 수 있습니다. ```python theme={null} import mlflow from mlflow.models import ModelSignature, infer_signature from mlflow.types.schema import Schema, ColSpec, ParamSchema, ParamSpec # 방법 1: 학습 데이터에서 자동 추론 (권장) import pandas as pd import numpy as np X_sample = pd.DataFrame({ "amount": [50000.0, 25.0], "merchant_category": ["online_retail", "grocery"], "hour": [3, 12], "is_international": [1, 0] }) y_sample = pd.DataFrame({ "fraud_probability": [0.85, 0.02], "is_fraud": [1, 0] }) # 입력/출력 데이터로 시그니처 자동 추론 signature = infer_signature(X_sample, y_sample) print(signature) # inputs: [amount: double, merchant_category: string, hour: long, is_international: long] # outputs: [fraud_probability: double, is_fraud: long] ``` ```python theme={null} # 방법 2: 수동으로 시그니처 정의 (정밀한 제어가 필요할 때) signature = ModelSignature( inputs=Schema([ ColSpec("double", "amount"), ColSpec("string", "merchant_category"), ColSpec("long", "hour"), ColSpec("integer", "is_international") ]), outputs=Schema([ ColSpec("double", "fraud_probability"), ColSpec("integer", "is_fraud"), ColSpec("string", "risk_level") ]), # params: 추론 시 동적으로 변경할 수 있는 파라미터 (선택) params=ParamSchema([ ParamSpec("threshold", "double", 0.7) ]) ) ``` ```python theme={null} # 방법 3: params를 활용한 동적 추론 파라미터 class FraudModelWithParams(mlflow.pyfunc.PythonModel): def load_context(self, context): import joblib self.model = joblib.load(context.artifacts["model_path"]) self.scaler = joblib.load(context.artifacts["scaler_path"]) def predict(self, context, model_input, params=None): # params에서 임계값을 런타임에 받음 (기본값 0.7) threshold = (params or {}).get("threshold", 0.7) scaled_input = self.scaler.transform(model_input) probabilities = self.model.predict_proba(scaled_input)[:, 1] return pd.DataFrame({ "fraud_probability": probabilities, "is_fraud": (probabilities >= threshold).astype(int) }) # params가 포함된 시그니처 생성 X_sample = X_test[:5] predictions = FraudModelWithParams().predict(None, X_sample) signature = infer_signature( X_sample, predictions, params={"threshold": 0.7} # 기본 파라미터 값 ) ``` > **시그니처 없이 배포하면** 입력 형식 오류가 서빙 환경에서 런타임에 발생합니다. `input_example`과 함께 `signature`를 항상 명시하는 것을 권장합니다. *** ## PyFunc 커스텀 모델 작성법 `mlflow.pyfunc.PythonModel`을 상속하면 `load_context()`와 `predict()` 메서드를 직접 구현하여 자유로운 추론 로직을 만들 수 있습니다. ### 기본 구조 ```python theme={null} import mlflow.pyfunc import pandas as pd import numpy as np class FraudDetectionModel(mlflow.pyfunc.PythonModel): """전처리 + 모델 추론 + 후처리를 하나로 묶는 커스텀 모델""" def load_context(self, context): """모델 로드 시 한 번 실행 (무거운 초기화 작업)""" import joblib # artifacts에서 모델과 전처리기 로드 self.model = joblib.load(context.artifacts["model_path"]) self.scaler = joblib.load(context.artifacts["scaler_path"]) self.threshold = 0.7 # 커스텀 임계값 def predict(self, context, model_input, params=None): """추론 요청마다 실행되는 메서드""" # 1. 전처리 scaled_input = self.scaler.transform(model_input) # 2. 예측 확률 계산 probabilities = self.model.predict_proba(scaled_input)[:, 1] # 3. 후처리 (커스텀 임계값 + 리스크 등급) results = pd.DataFrame({ "fraud_probability": probabilities, "is_fraud": (probabilities >= self.threshold).astype(int), "risk_level": pd.cut( probabilities, bins=[0, 0.3, 0.7, 1.0], labels=["LOW", "MEDIUM", "HIGH"] ) }) return results ``` ### 커스텀 모델 저장 및 등록 ```python theme={null} import joblib from sklearn.ensemble import GradientBoostingClassifier from sklearn.preprocessing import StandardScaler # 모델과 스케일러 학습 scaler = StandardScaler() X_train_scaled = scaler.fit_transform(X_train) model = GradientBoostingClassifier(n_estimators=200) model.fit(X_train_scaled, y_train) # 아티팩트 저장 joblib.dump(model, "/tmp/model.joblib") joblib.dump(scaler, "/tmp/scaler.joblib") # 시그니처 자동 추론 sample_input = pd.DataFrame(X_test[:3], columns=feature_cols) sample_output = FraudDetectionModel().predict(None, sample_input) signature = infer_signature(sample_input, sample_output) # 커스텀 모델을 MLflow에 로깅 with mlflow.start_run(run_name="fraud-custom-pyfunc"): model_info = mlflow.pyfunc.log_model( artifact_path="fraud_model", python_model=FraudDetectionModel(), artifacts={ "model_path": "/tmp/model.joblib", "scaler_path": "/tmp/scaler.joblib" }, conda_env={ "dependencies": [ "python=3.10", {"pip": ["scikit-learn==1.4.0", "pandas>=2.0", "numpy>=1.24"]} ] }, signature=signature, input_example=sample_input, registered_model_name="catalog.schema.fraud_detection_custom" ) ``` > **의존성 관리 주의**: `conda_env` 또는 `pip_requirements`를 명시하지 않으면, 로컬 환경의 패키지 버전이 서빙 환경과 달라 오류가 발생할 수 있습니다. 반드시 학습에 사용한 주요 라이브러리 버전을 명시하십시오.