matalee/hpce-dev

1from __future__ import annotations2"""3Model-based Intent Inference ([2b] 모듈)4 5시나리오 무관 sklearn Logistic Regression 추론 머신러리.6- 학습 데이터(training_data)·dataset_path·model_prefix는 호출자(시나리오 엔진)가 주입7- StandardScaler + LogisticRegression Pipeline8- MLflow Registry 등록 (모델명: {model_prefix}{intent_id}_sklearn)9- seed 고정 (42)으로 재현성 확보10"""11import json12import logging13import random14from pathlib import Path15from typing import Any16 17import numpy as np18import mlflow19import mlflow.sklearn20from sklearn.linear_model import LogisticRegression21from sklearn.preprocessing import StandardScaler22from sklearn.pipeline import Pipeline23 24from config import settings25 26logger = logging.getLogger(__name__)27_model_cache: dict[str, Any] = {}28 29 30# ── Public API ────────────────────────────────────────────────31 32def _train_pipeline(X: list, y: list, seed: int = 42, train_params: dict | None = None) -> Pipeline:33    """StandardScaler + LogisticRegression 파이프라인을 학습한다.34 35    seed를 고정하여 재현성을 확보한다.36 37    Args:38        X: 특징 행렬.39        y: 레이블 벡터.40        seed: 난수 시드 (재현성).41        train_params: 시나리오 config L2.model.train의 하이퍼파라미터.42            class_weight·C를 override 가능 (기본 balanced·C=1.0).43 44    Returns:45        학습된 sklearn Pipeline.46    """47    tp = train_params or {}48    pipe = Pipeline([49        ("scaler", StandardScaler()),50        ("lr", LogisticRegression(51            random_state=seed,52            max_iter=500,53            C=tp.get("C", 1.0),54            class_weight=tp.get("class_weight", "balanced"),55        )),56    ])57    pipe.fit(np.array(X, dtype=float), np.array(y))58    return pipe59 60 61def _extract_from_dataset(62    intent_id: str,63    feature_names: list[str],64    seed: int,65    dataset_path: Path,66    neg_pos_ratio: float = 2.0,67) -> tuple[list[list[float]], list[int]] | None:68    """seed_dataset.json에서 intent_id에 대한 (X, y)를 추출한다.69 70    양성은 sample["intent_labels"]에 intent_id가 있는 경우(y=1), 음성은 그 외(y=0)이다.71    클래스 불균형 처리를 위해 음성은 neg_pos_ratio × n_pos 까지만 샘플링한다.72 73    Args:74        intent_id: 추출할 Intent ID.75        feature_names: feature 벡터를 구성할 feature 이름 순서.76        seed: 음성 샘플링에 사용할 난수 시드.77        dataset_path: seed_dataset.json 경로.78        neg_pos_ratio: 양성 대비 음성 샘플 비율 상한.79 80    Returns:81        (X, y) 튜플. 양성·음성이 각각 3건 미만이거나 데이터셋이 없으면 None.82    """83    if not dataset_path.exists():84        return None85    try:86        with open(dataset_path, encoding="utf-8") as f:87            dataset = json.load(f)88    except Exception as e:89        logger.warning(f"Failed to load seed_dataset.json: {e}")90        return None91 92    X_pos, X_neg = [], []93    for sample in dataset.get("samples", []):94        # batch + pattern + event 를 합친 전체 feature 벡터 (추론 시점과 동일 공간)95        feats = {96            **sample.get("batch_features", {}),97            **sample.get("pattern_features", {}),98            **sample.get("event_features", {}),99        }100        x = [float(feats.get(name, 0.0)) for name in feature_names]101        if intent_id in sample.get("intent_labels", {}):102            X_pos.append(x)103        else:104            X_neg.append(x)105 106    if len(X_pos) < 3 or len(X_neg) < 3:107        return None108 109    rng = random.Random(seed)110    n_neg_target = min(len(X_neg), max(int(len(X_pos) * neg_pos_ratio), 10))111    X_neg_sampled = rng.sample(X_neg, n_neg_target) if len(X_neg) > n_neg_target else X_neg112 113    X = X_pos + X_neg_sampled114    y = [1] * len(X_pos) + [0] * len(X_neg_sampled)115    return X, y116 117 118def train_and_register(119    intent_id: str,120    training_data: dict,121    dataset_path: Path,122    model_prefix: str,123    seed: int = 42,124    train_params: dict | None = None,125) -> Pipeline | None:126    """Intent의 학습 데이터로 모델을 학습하고 MLflow에 등록한다.127 128    데이터 소스 우선순위:129      1) dataset_path(seed_dataset.json)의 페르소나 시드 데이터셋130      2) training_data[intent_id]의 도메인 지식 X, y131 132    Args:133        intent_id: 학습할 Intent ID.134        training_data: Intent별 학습 정의(features·X·y 등).135        dataset_path: seed_dataset.json 경로.136        model_prefix: 시나리오별 MLflow 모델명 네임스페이스.137            등록명은 {model_prefix}{intent_id}_sklearn.138        seed: 난수 시드 (재현성).139        train_params: 학습 하이퍼파라미터(class_weight/C).140 141    Returns:142        학습된 Pipeline. 학습 데이터가 없으면 None.143    """144    data = training_data.get(intent_id)145    if data is None:146        return None147 148    feature_names = data["features"]149    model_name = f"{model_prefix}{intent_id}_sklearn"150 151    # 1) 시드 데이터셋 우선152    extracted = _extract_from_dataset(intent_id, feature_names, seed, dataset_path)153    if extracted is not None:154        X, y = extracted155        data_source = "seed_dataset"156    elif "X" in data and "y" in data:157        X, y = data["X"], data["y"]158        data_source = "domain_knowledge"159    else:160        return None161 162    pipe = _train_pipeline(X, y, seed=seed, train_params=train_params)163 164    mlflow.set_tracking_uri(settings.MLFLOW_URI)165    with mlflow.start_run(run_name=f"{model_name}_init"):166        mlflow.sklearn.log_model(167            pipe,168            "model",169            registered_model_name=model_name,170        )171        mlflow.log_params({172            "intent_id":     intent_id,173            "n_features":    len(feature_names),174            "n_samples":     len(y),175            "n_positive":    int(sum(y)),176            "seed":          seed,177            "data_source":   data_source,178            "feature_names": ",".join(feature_names),179        })180        train_acc = pipe.score(np.array(X, dtype=float), np.array(y))181        mlflow.log_metric("train_accuracy", train_acc)182 183    logger.info(f"Trained + registered: {model_name} "184                f"(source={data_source}, n={len(y)}, pos={int(sum(y))}, acc={train_acc:.3f})")185    return pipe186 187 188def _load_or_train(189    intent_id: str,190    training_data: dict,191    dataset_path: Path,192    model_prefix: str,193    train_params: dict | None = None,194) -> Pipeline | None:195    """모델을 캐시→MLflow Registry 순으로 로드하고, 없으면 학습·등록한다.196 197    프로세스 캐시(_model_cache)를 사용해 intent별로 1회만 로드/학습한다.198 199    Args:200        intent_id: 로드/학습할 Intent ID.201        training_data: Intent별 학습 정의.202        dataset_path: seed_dataset.json 경로.203        model_prefix: 시나리오별 MLflow 모델명 네임스페이스.204        train_params: 학습 하이퍼파라미터.205 206    Returns:207        로드 또는 학습된 Pipeline. 학습 정의가 없으면 None.208    """209    cache_key = f"{model_prefix}{intent_id}"210    if cache_key in _model_cache:211        return _model_cache[cache_key]212 213    if intent_id not in training_data:214        return None215 216    mlflow.set_tracking_uri(settings.MLFLOW_URI)217    uri = f"models:/{model_prefix}{intent_id}_sklearn/latest"218    try:219        pipe = mlflow.sklearn.load_model(uri)220    except Exception:221        pipe = train_and_register(222            intent_id, training_data=training_data,223            dataset_path=dataset_path, model_prefix=model_prefix,224            train_params=train_params,225        )226 227    _model_cache[cache_key] = pipe228    return pipe229 230 231def predict(232    intent_id: str,233    features: dict[str, Any],234    training_data: dict,235    dataset_path: Path,236    model_prefix: str,237    train_params: dict | None = None,238) -> float:239    """Intent ID에 대해 Model 기반 Score를 추론한다.240 241    features dict에서 학습에 사용된 피처들을 순서대로 추출하며,242    누락된 피처는 0.0으로 처리한다.243 244    Args:245        intent_id: 추론할 Intent ID.246        features: 추론에 사용할 feature dict.247        training_data: Intent별 학습 정의(시나리오 엔진 제공).248        dataset_path: seed_dataset.json 경로(시나리오 엔진 제공).249        model_prefix: 시나리오별 모델명 네임스페이스(시나리오 엔진 제공).250        train_params: 학습 하이퍼파라미터(class_weight/C, config L2.model.train).251 252    Returns:253        0~1 범위의 예측 점수. 모델이 없으면 0.0.254    """255    pipe = _load_or_train(intent_id, training_data, dataset_path, model_prefix, train_params)256    if pipe is None:257        return 0.0258 259    feature_names = training_data[intent_id]["features"]260    x = np.array([[float(features.get(name, 0.0)) for name in feature_names]])261 262    proba = pipe.predict_proba(x)[0][1]263    return float(proba)264 265 266def explain(267    intent_id: str,268    features: dict[str, Any],269    training_data: dict,270    dataset_path: Path,271    model_prefix: str,272    top: int = 3,273) -> list[dict]:274    """Model 추론의 feature 기여도를 분해한다.275 276    선형 파이프라인(StandardScaler + LogisticRegression)에서277    기여_i = coef_i × ((x_i - mean_i) / scale_i)로 계산하고,278    |기여| 상위 top개를 반환한다.279 280    Args:281        intent_id: 기여도를 분해할 Intent ID.282        features: 추론에 사용한 feature dict.283        training_data: Intent별 학습 정의.284        dataset_path: seed_dataset.json 경로.285        model_prefix: 시나리오별 모델명 네임스페이스.286        top: 반환할 상위 기여 feature 개수.287 288    Returns:289        feature별 기여 정보(label·contribution·direction·value) dict의 목록.290        모델이 없거나 분해에 실패하면 빈 목록.291    """292    pipe = _load_or_train(intent_id, training_data, dataset_path, model_prefix)293    if pipe is None or intent_id not in training_data:294        return []295    feats = training_data[intent_id]["features"]296    x = np.array([float(features.get(n, 0.0)) for n in feats])297    try:298        scaler = pipe.named_steps["scaler"]299        lr = pipe.named_steps["lr"]300        xs = (x - scaler.mean_) / scaler.scale_301        contrib = lr.coef_[0] * xs302    except Exception:303        return []304    items = sorted(zip(feats, contrib, x), key=lambda t: -abs(t[1]))[:top]305    return [{"label": n, "contribution": round(float(c), 4),306             "direction": "up" if c >= 0 else "down", "value": round(float(v), 2)}307            for n, c, v in items]308 309 310def train_all(311    training_data: dict,312    dataset_path: Path,313    model_prefix: str,314    seed: int = 42,315) -> dict[str, float]:316    """training_data의 모든 Model Intent를 학습·등록한다.317 318    Args:319        training_data: Intent별 학습 정의.320        dataset_path: seed_dataset.json 경로.321        model_prefix: 시나리오별 MLflow 모델명 네임스페이스.322        seed: 난수 시드 (재현성).323 324    Returns:325        학습에 성공한 Intent에 대한 {intent_id: 1.0} 매핑.326    """327    results = {}328    for intent_id in training_data.keys():329        pipe = train_and_register(330            intent_id, training_data=training_data, seed=seed,331            dataset_path=dataset_path, model_prefix=model_prefix,332        )333        if pipe is not None:334            results[intent_id] = 1.0335    return results336