matalee/hpce-dev

1from __future__ import annotations2"""Intent Inference 통합.3 4흐름:5  - infer_batch(survey)  : Batch Feature만으로 baseline Intent Score 산출6  - infer_with_behavior  : Batch + Behavioral Pattern Feature를 합쳐 재추론7                          (boost 누적 방식이 아니라, 모든 피처를 입력으로 다시 모델/룰을 통과시킴)8 9산출되는 IntentScore는 baseline 대비 변화량(delta_score, rank_change)을 함께 보관한다.10"""11import logging12import math13from dataclasses import dataclass14from typing import Any15 16# Softmax 분포 sharpness 조절. 작을수록 Top 점수에 분포가 집중.17# T=0.15 → 시연 임팩트(상위 Intent 강조) 우선. Top 5가 분포의 ~60% 점유, 행동 1번에 Δp 수%p18PROBABILITY_TEMPERATURE = 0.1519 20from config import settings21from core.engines import get_engine, config22from core.engines.base import ScenarioEngine23from core.extractor import get_extractor24 25logger = logging.getLogger(__name__)26 27# 행동 → 직접 신호 Intent 부스트. (시나리오 config L2_inference.ranker.action_signal로 override)28# Rule pattern_boost가 일부 행동만 커버하는 한계를 보완해, behaviors.json의 모든 행동이29# 의미상 연결된 Intent를 끌어올리도록 한다. (final 재추론에만 적용; baseline은 batch만)30# 아래 값은 scenario_id 미지정/누락 시 fallback 기본값.31ACTION_SIGNAL_SCALE = 0.28   # 최신 행동 1회당 가산 (weight=1.0 기준)32ACTION_SIGNAL_CAP   = 0.55   # 행동 반복 시 상한33ACTION_SIGNAL_DECAY = 0.6    # 위치 기반 recency 감쇠 (최신 age=0 → 1.0, 직전 0.6, 그전 0.36 …)34 35 36def _resolve_temperature(scenario_id: str | None) -> float:37    """시나리오 config의 softmax 온도를 조회한다.38 39    Args:40        scenario_id: 조회할 시나리오 ID. None이면 모듈 기본값을 사용한다.41 42    Returns:43        softmax 온도. config 누락 시 모듈 기본값으로 fallback한다.44    """45    if scenario_id is None:46        return PROBABILITY_TEMPERATURE47    try:48        return config.get_probability_temperature(scenario_id)49    except (KeyError, FileNotFoundError):50        return PROBABILITY_TEMPERATURE51 52 53def _resolve_action_signal(scenario_id: str | None) -> tuple[float, float, float]:54    """시나리오 config의 행동 부스트 파라미터를 조회한다.55 56    Args:57        scenario_id: 조회할 시나리오 ID. None이면 모듈 기본값을 사용한다.58 59    Returns:60        (scale, cap, decay) 튜플. config 누락 시 모듈 기본값으로 fallback한다.61    """62    if scenario_id is not None:63        try:64            sig = config.get_action_signal(scenario_id)65            return sig["scale"], sig["cap"], sig["decay"]66        except (KeyError, FileNotFoundError):67            pass68    return ACTION_SIGNAL_SCALE, ACTION_SIGNAL_CAP, ACTION_SIGNAL_DECAY69 70 71def _resolve_boost_mode(scenario_id: str | None) -> str:72    """행동 boost 합성 방식을 조회한다.73 74    config L2.ranker.action_signal.boost_mode 값을 사용한다.75 76    Args:77        scenario_id: 조회할 시나리오 ID. None이면 기본값을 사용한다.78 79    Returns:80        합성 방식 문자열. 누락 시 'additive'로 fallback한다.81    """82    if scenario_id is not None:83        try:84            return config.get_action_signal(scenario_id).get("boost_mode", "additive")85        except (KeyError, FileNotFoundError):86            pass87    return "additive"88 89 90def _resolve_action_suppress(scenario_id: str | None) -> dict | None:91    """행동 기반 의도 감쇠 설정을 조회한다.92 93    config L2.ranker.action_signal.suppress 값을 사용한다.94    형식: {"by_entity": {entity: [감쇠 대상 intent_id, ...]}, "scale": float, "cap": float}.95    예) 회복 행동(mental_recovery/exercise) 누적 시 번아웃 심화 intent를 점진 감쇠.96 97    Args:98        scenario_id: 조회할 시나리오 ID. None이면 None을 반환한다.99 100    Returns:101        감쇠 설정 dict. 미설정 시 None.102    """103    if scenario_id is not None:104        try:105            return config.get_action_signal(scenario_id).get("suppress")106        except (KeyError, FileNotFoundError):107            pass108    return None109 110 111def _action_intent_signals(112    events: list[dict],113    behavior_map: dict[str, list[str]],114    decay: float = ACTION_SIGNAL_DECAY,115) -> dict[str, float]:116    """세션 누적 행동을 entity→intent 매핑으로 의도별 가중 신호로 환산한다.117 118    - recency decay: 최신 행동일수록 큰 weight (DECAY^age). 방금 한 행동이 현재 의도를 주도하되,119      같은 행동 반복은 누적되어 강해진다(과거도 0으로 죽이진 않음).120    - BACK(navigate_back)은 메뉴 복귀용 순수 내비게이션 → 신호·aging 모두에서 제외(무효과).121      섹션을 떠난 행동은 이후 다른 행동이 쌓이며 decay로 자연 소멸한다.122 123    Args:124        events: 세션에 누적된 행동 이벤트 리스트.125        behavior_map: entity → intent_id 리스트 매핑.126        decay: 위치 기반 recency 감쇠 계수.127 128    Returns:129        intent_id → 가중 신호 합 매핑.130    """131    real = [ev for ev in events if ev.get("event_type") != "navigate_back"]132    n = len(real)133 134    weights: dict[str, float] = {}135    for i, ev in enumerate(real):136        age = (n - 1) - i137        w = decay ** age138        for iid in behavior_map.get(ev.get("entity", ""), []):139            weights[iid] = weights.get(iid, 0.0) + w140    return weights141 142 143@dataclass144class IntentScore:145    """단일 intent의 baseline·final 점수와 순위 변화(행동 반영 전후)."""146    intent_id:       str147    intent_name:     str148    L1_id:           str149    L1_name:         str150    L2_id:           str151    L2_name:         str152    inference_type:  str153    baseline_score:  float   # Batch Feature만으로 추론한 점수154    final_score:     float   # Batch + Behavioral Pattern Feature 합쳐 재추론한 점수155    delta_score:     float   # final - baseline156    baseline_rank:   int     # baseline 기준 rank157    rank:            int     # final_score 기준 rank158    rank_change:     int     # baseline_rank - rank (양수면 상승)159 160 161def infer_batch(162    survey_answers: dict[str, str],163    scenario_id: str = settings.SCENARIO_ID,164) -> tuple[dict[str, Any], list[IntentScore]]:165    """설문 답변만으로 baseline Intent Score를 산출한다 (행동 반영 전).166 167    Args:168        survey_answers: 질문 ID → 선택 응답 코드 매핑.169        scenario_id: 추론에 사용할 시나리오 ID.170 171    Returns:172        (batch_features, scores) 튜플. batch_features는 산출된 Batch Feature,173        scores는 final 점수 내림차순으로 정렬된 IntentScore 리스트.174    """175    engine = get_engine(scenario_id)176    batch_features = engine.build_batch_features(survey_answers)177    all_features = {178        **batch_features,179        **engine.empty_pattern_features(),180        **engine.empty_event_features(),181    }182 183    raw = _score_all(all_features, engine)184    scores = _to_intent_scores(raw, raw, engine)185    return batch_features, scores186 187 188def infer_with_behavior(189    survey_answers: dict[str, str],190    session_id: str,191    scenario_id: str = settings.SCENARIO_ID,192) -> tuple[dict[str, Any], list[IntentScore]]:193    """Batch + 누적 Pattern + 최신 Event Feature를 합쳐 재추론한다.194 195    baseline(행동 없는 상태) 점수를 함께 산출해 delta_score / rank_change를 채운다.196 197    Args:198        survey_answers: 질문 ID → 선택 응답 코드 매핑.199        session_id: 누적 행동 이벤트를 조회할 세션 ID.200        scenario_id: 추론에 사용할 시나리오 ID.201 202    Returns:203        (batch_features, scores) 튜플. scores는 baseline 대비 델타·순위변화가 채워진204        IntentScore 리스트로, final 점수 내림차순으로 정렬된다.205    """206    engine = get_engine(scenario_id)207    batch_features = engine.build_batch_features(survey_answers)208 209    # baseline: Pattern/Event Feature를 0으로 둔 상태210    baseline_features = {211        **batch_features,212        **engine.empty_pattern_features(),213        **engine.empty_event_features(),214    }215    baseline_raw = _score_all(baseline_features, engine)216 217    # final: 실제 누적 Pattern + 최신 Event Feature 반영 (엔진 전용 계산)218    pattern_features = engine.pattern_features(session_id)219    event_features = engine.event_features(session_id)220    events = get_extractor()._events_by_session.get(session_id, [])221 222    combined_features = {**batch_features, **pattern_features, **event_features}223    final_raw = _score_all(combined_features, engine)224 225    # 행동이 직접 가리키는 Intent를 끌어올림 (final 에만 적용 → Δ·rank_change가 행동에 귀속)226    scale, cap, decay = _resolve_action_signal(scenario_id)227    boost_mode = _resolve_boost_mode(scenario_id)   # "additive"(기본) | "headroom"228    behavior_map = engine.behavior_intent_map()229    for iid, cnt in _action_intent_signals(events, behavior_map, decay=decay).items():230        if iid in final_raw:231            boost = min(cnt * scale, cap)232            if boost_mode == "headroom":233                # 헤드룸 비례 가산: 천장(0.97) 다중 동점 방지 + base 순서 보존234                final_raw[iid] = final_raw[iid] + boost * (1.0 - final_raw[iid])235            else:236                final_raw[iid] = min(final_raw[iid] + boost, 0.97)237 238    # 행동 기반 감쇠: 특정 행동(예: 회복) 누적 시 대상 intent(예: 번아웃 심화)를 점진 감쇠.239    # 가산 boost와 대칭 — 같은 recency decay 가중을 penalty로 환산해 점수를 비율 축소.240    suppress = _resolve_action_suppress(scenario_id)241    if suppress:242        sup_scale = suppress.get("scale", 0.0)243        sup_cap   = suppress.get("cap", 1.0)244        sup_map   = suppress.get("by_entity", {})245        for iid, cnt in _action_intent_signals(events, sup_map, decay=decay).items():246            if iid in final_raw:247                penalty = min(cnt * sup_scale, sup_cap)248                final_raw[iid] = final_raw[iid] * (1.0 - penalty)249 250    scores = _to_intent_scores(baseline_raw, final_raw, engine)251    return batch_features, scores252 253 254def _score_all(features: dict[str, Any], engine: ScenarioEngine) -> dict[str, float]:255    """모든 Intent에 대해 점수만 산출한다.256 257    inference_type에 따라 rule/model로 분기한다.258 259    Args:260        features: 추론 입력 feature 매핑.261        engine: 시나리오 엔진.262 263    Returns:264        intent_id → score 매핑.265    """266    f = dict(features)267    if isinstance(f.get("결합 여부"), bool):268        f["결합 여부"] = 1 if f["결합 여부"] else 0269 270    out: dict[str, float] = {}271    for intent in engine.intents():272        iid = intent["id"]273        if intent["inference_type"] == "Model":274            score = engine.model_predict(iid, f)275        else:276            score = engine.rule_predict(iid, f)277        out[iid] = float(score)278    return out279 280 281def _rank_map(raw: dict[str, float]) -> dict[str, int]:282    """raw 점수를 내림차순 순위로 환산한다.283 284    Args:285        raw: intent_id → score 매핑.286 287    Returns:288        intent_id → 순위(1-기반) 매핑.289    """290    ordered = sorted(raw.items(), key=lambda kv: kv[1], reverse=True)291    return {iid: i for i, (iid, _) in enumerate(ordered, start=1)}292 293 294def _to_intent_scores(295    baseline_raw: dict[str, float],296    final_raw: dict[str, float],297    engine: ScenarioEngine,298) -> list[IntentScore]:299    """baseline·final raw 점수를 IntentScore 리스트로 변환한다.300 301    델타·순위변화를 채우고 final 점수 내림차순으로 정렬한다.302 303    Args:304        baseline_raw: baseline intent_id → score 매핑.305        final_raw: final intent_id → score 매핑.306        engine: 시나리오 엔진.307 308    Returns:309        final 점수 내림차순으로 정렬된 IntentScore 리스트.310    """311    baseline_ranks = _rank_map(baseline_raw)312    final_ranks    = _rank_map(final_raw)313 314    results: list[IntentScore] = []315    for intent in engine.intents():316        iid = intent["id"]317        b = baseline_raw.get(iid, 0.0)318        f = final_raw.get(iid, 0.0)319        br = baseline_ranks.get(iid, 0)320        fr = final_ranks.get(iid, 0)321        results.append(IntentScore(322            intent_id=iid,323            intent_name=intent["name"],324            L1_id=intent["L1_id"],325            L1_name=intent["L1_name"],326            L2_id=intent["L2_id"],327            L2_name=intent["L2_name"],328            inference_type=intent["inference_type"],329            baseline_score=round(b, 4),330            final_score=round(f, 4),331            delta_score=round(f - b, 4),332            baseline_rank=br,333            rank=fr,334            rank_change=br - fr,335        ))336 337    results.sort(key=lambda s: s.final_score, reverse=True)338    return results339 340 341# ── WebSocket 페이로드 헬퍼 ───────────────────────────────────342 343def _softmax(values: list[float], temperature: float) -> list[float]:344    """수치적으로 안정적인 softmax로 raw score를 정규화 확률 분포로 변환한다.345 346    Args:347        values: 정규화할 raw score 리스트.348        temperature: softmax 온도. 작을수록 상위 값에 분포가 집중된다.349 350    Returns:351        합이 1이 되는 정규화 확률 리스트. 입력이 비면 빈 리스트.352    """353    if not values:354        return []355    scaled = [v / temperature for v in values]356    m = max(scaled)357    exps = [math.exp(s - m) for s in scaled]358    total = sum(exps) or 1.0359    return [e / total for e in exps]360 361 362def to_probability_dict(363    scores: list[IntentScore],364    scenario_id: str | None = None,365    temperature: float | None = None,366) -> dict[str, dict[str, float]]:367    """Intent raw score를 softmax 정규화 확률(p)과 baseline 확률(p0)로 변환한다.368 369    raw score 합 분모 정규화는 분포가 너무 평탄하므로 softmax(score / T) 분포를 사용한다.370    T가 작을수록 상위 Intent에 분포가 집중된다.371    T는 scenario_id의 config(L2_inference.calibrator)에서 조회한다(미지정 시 모듈 기본값).372 373    INTENT_UPDATE 페이로드의 `all_probabilities` 필드에 사용된다.374 375    Args:376        scores: 확률로 변환할 IntentScore 리스트.377        scenario_id: 온도 조회에 사용할 시나리오 ID. None이면 모듈 기본값.378        temperature: softmax 온도 직접 지정. None이면 scenario_id로 조회한다.379 380    Returns:381        intent_id → {"p": final 확률, "p0": baseline 확률} 매핑.382    """383    if temperature is None:384        temperature = _resolve_temperature(scenario_id)385    p_vals  = _softmax([s.final_score    for s in scores], temperature)386    p0_vals = _softmax([s.baseline_score for s in scores], temperature)387    return {388        s.intent_id: {389            "p":  round(p_vals[i],  6),390            "p0": round(p0_vals[i], 6),391        }392        for i, s in enumerate(scores)393    }394 395 396def to_topn_with_others(397    scores: list[IntentScore],398    top_n: int = 5,399    scenario_id: str | None = None,400) -> tuple[list[dict], dict]:401    """Top-N + 기타(others) 페이로드를 구성한다.402 403    Args:404        scores: 확률로 변환할 IntentScore 리스트.405        top_n: 상위로 노출할 intent 개수.406        scenario_id: 온도 조회에 사용할 시나리오 ID. None이면 모듈 기본값.407 408    Returns:409        (top_list, others) 튜플.410        top_list: [ {intent_id, ..., probability, baseline_probability, delta_probability} ]411        others: { count, probability, baseline_probability, delta_probability }412    """413    probs = to_probability_dict(scores, scenario_id=scenario_id)414    sorted_scores = sorted(scores, key=lambda s: s.final_score, reverse=True)415 416    top_items: list[dict] = []417    for s in sorted_scores[:top_n]:418        pr = probs[s.intent_id]419        top_items.append({420            "intent_id":            s.intent_id,421            "intent_nm_ko":         s.intent_name,422            "L1_id":                s.L1_id,423            "L1_name":              s.L1_name,424            "L2_id":                s.L2_id,425            "L2_name":              s.L2_name,426            "inference_type":       s.inference_type,427            "rank":                 s.rank,428            "baseline_rank":        s.baseline_rank,429            "rank_change":          s.rank_change,430            "probability":          pr["p"],431            "baseline_probability": pr["p0"],432            "delta_probability":    round(pr["p"] - pr["p0"], 6),433        })434 435    rest = sorted_scores[top_n:]436    others_p  = sum(probs[s.intent_id]["p"]  for s in rest)437    others_p0 = sum(probs[s.intent_id]["p0"] for s in rest)438    others = {439        "count":                len(rest),440        "probability":          round(others_p,  6),441        "baseline_probability": round(others_p0, 6),442        "delta_probability":    round(others_p - others_p0, 6),443    }444    return top_items, others445 446 447# ── Customer Context JSON 생성 ────────────────────────────────448 449def to_customer_context_json(450    session_id: str,451    stage: str,452    scenario_id: str,453    scores: list[IntentScore],454    batch_features: dict[str, Any],455) -> dict[str, Any]:456    """세션·단계·전체 intent 점수를 Customer Context JSON으로 직렬화한다.457 458    저장/조회용 페이로드를 구성한다.459 460    Args:461        session_id: 세션 ID.462        stage: 추론 단계.463        scenario_id: 시나리오 ID.464        scores: 직렬화할 IntentScore 리스트.465        batch_features: 산출된 Batch Feature 매핑.466 467    Returns:468        session/scenario/stage/intents 키를 가진 Customer Context JSON dict.469    """470    intents = []471    for s in scores:472        intents.append({473            "intent_id":         s.intent_id,474            "intent_nm_ko":      s.intent_name,475            "L1":                {"id": s.L1_id, "name": s.L1_name},476            "L2":                {"id": s.L2_id, "name": s.L2_name},477            "baseline_score":    s.baseline_score,478            "final_score":       s.final_score,479            "delta_score":       s.delta_score,480            "baseline_rank":     s.baseline_rank,481            "rank":              s.rank,482            "rank_change":       s.rank_change,483            "inference_type":    s.inference_type,484        })485 486    return {487        "session_id":       session_id,488        "scenario_id":      scenario_id,489        "stage":            stage,490        "intents":          intents,491    }492