import gradio as gr import numpy as np import pandas as pd from sklearn.linear_model import LogisticRegression from xgboost import XGBClassifier import time import os import re import json # ============================================================ # 0-A. Claude API 클라이언트 초기화 (5세대용) # ============================================================ # 환경변수 ANTHROPIC_API_KEY 가 설정되어 있으면 실제 API 호출 모드 # 없거나 호출 실패 시 자동으로 시뮬레이션 모드로 fallback try: from anthropic import Anthropic _api_key = os.environ.get("ANTHROPIC_API_KEY") if _api_key: claude_client = Anthropic(api_key=_api_key) CLAUDE_AVAILABLE = True else: claude_client = None CLAUDE_AVAILABLE = False except ImportError: claude_client = None CLAUDE_AVAILABLE = False CLAUDE_MODEL = "claude-sonnet-4-6" # 2026년 5월 현재 권장 모델 # ============================================================ # 0-B. Mini GNN (4세대) - numpy 순수 구현 + 진짜 학습 # ============================================================ # 250건 데이터로 실제 학습되는 작은 GNN # 구조: 4차원 입력 → 은닉층1 (16D) → 은닉층2 (16D) → 사기 확률 # 메시지 패싱을 2번 수행 (2-hop) def build_training_data_v2(n_normal=200, n_fraud=50, seed=42): """4세대 GNN용 학습 데이터 (250건, 1-3세대와 동일 분포·동일 규모).""" np.random.seed(seed) normal_arr = np.column_stack([ np.random.normal(100, 50, n_normal), np.random.normal(14, 4, n_normal), np.random.binomial(1, 0.1, n_normal), np.random.normal(1, 0.5, n_normal), ]) fraud_arr = np.column_stack([ np.random.normal(500, 200, n_fraud), np.random.normal(3, 2, n_fraud), np.random.binomial(1, 0.8, n_fraud), np.random.normal(10, 5, n_fraud), ]) X = np.vstack([normal_arr, fraud_arr]) y = np.concatenate([np.zeros(n_normal), np.ones(n_fraud)]) return X, y def build_graph_features(amount, hour, new_payee, ratio): """ 단일 거래로부터 4×4 노드 임베딩 행렬을 생성한다. - 노드 0: 거래 자체 (분류 대상) - 노드 1: 송금인 - 노드 2: 수취인 (사기 시그널 포함) - 노드 3: 단말기 """ amount_n = (amount - 100) / 200 hour_n = (hour - 14) / 8 ratio_n = (ratio - 1) / 5 trans_node = np.array([amount_n, hour_n, new_payee, ratio_n]) sender_node = np.array([0.0, hour_n, 0.0, 0.0]) if new_payee == 1: receiver_node = np.array([0.5, 0.3, 1.0, 0.4]) else: receiver_node = np.array([-0.2, 0.0, 0.0, -0.1]) if new_payee == 1 and (hour <= 6 or hour >= 22): device_node = np.array([0.3, 0.5, 0.0, 0.2]) else: device_node = np.array([0.0, 0.0, 0.0, 0.0]) return np.array([trans_node, sender_node, receiver_node, device_node]) # Adjacency matrix (정규화) ADJ = np.array([ [1, 1, 1, 1], # 거래 노드: 자기 + 모든 이웃과 연결 [1, 1, 0, 0], # 송금인: 자기 + 거래 [1, 0, 1, 0], # 수취인: 자기 + 거래 [1, 0, 0, 1], # 단말기: 자기 + 거래 ], dtype=np.float32) ADJ_NORM = ADJ / ADJ.sum(axis=1, keepdims=True) class MiniGNN: """순수 numpy로 구현된 2-layer GNN + MLP.""" HIDDEN_DIM = 16 def __init__(self, seed=42): np.random.seed(seed) self.W1 = np.random.randn(4, self.HIDDEN_DIM) * np.sqrt(2.0 / 4) self.W2 = np.random.randn(self.HIDDEN_DIM, self.HIDDEN_DIM) * np.sqrt(2.0 / self.HIDDEN_DIM) self.W_mlp = np.random.randn(self.HIDDEN_DIM, 1) * np.sqrt(2.0 / self.HIDDEN_DIM) self.b_mlp = np.zeros(1) @staticmethod def _relu(x): return np.maximum(0, x) @staticmethod def _relu_grad(x): return (x > 0).astype(np.float32) @staticmethod def _sigmoid(x): return 1 / (1 + np.exp(-np.clip(x, -50, 50))) def forward(self, node_features, return_intermediates=False, adj_mask=None): """forward pass. adj_mask가 주어지면 ADJ_NORM 대신 사용 (GNNExplainer용 엣지 마스킹). """ adj = adj_mask if adj_mask is not None else ADJ_NORM agg1 = adj @ node_features z1 = agg1 @ self.W1 h1 = self._relu(z1) agg2 = adj @ h1 z2 = agg2 @ self.W2 h2 = self._relu(z2) trans_embedding = h2[0] logit = trans_embedding @ self.W_mlp + self.b_mlp prob = self._sigmoid(logit) if return_intermediates: return float(prob[0]), { 'h1': h1, 'h2': h2, 'z1': z1, 'z2': z2, 'agg1': agg1, 'agg2': agg2, 'trans_embedding': trans_embedding, 'logit': float(logit[0]), } return float(prob[0]) def train_step(self, node_features, label, lr=0.05): prob, cache = self.forward(node_features, return_intermediates=True) dlogit = (prob - label) dW_mlp = cache['trans_embedding'].reshape(-1, 1) * dlogit db_mlp = np.array([dlogit]) dh2 = np.zeros_like(cache['h2']) dh2[0] = self.W_mlp.flatten() * dlogit dz2 = dh2 * self._relu_grad(cache['z2']) dW2 = cache['agg2'].T @ dz2 dh1 = (ADJ_NORM.T @ dz2) @ self.W2.T dz1 = dh1 * self._relu_grad(cache['z1']) dW1 = cache['agg1'].T @ dz1 self.W1 -= lr * dW1 self.W2 -= lr * dW2 self.W_mlp -= lr * dW_mlp self.b_mlp -= lr * db_mlp def train_gnn(): """앱 시작 시 1회 실행. 약 2초 소요.""" X, y = build_training_data_v2() graphs = np.array([ build_graph_features(X[i, 0], X[i, 1], X[i, 2], X[i, 3]) for i in range(len(X)) ]) model = MiniGNN(seed=42) np.random.seed(123) for epoch in range(100): indices = np.random.permutation(len(X)) for i in indices: model.train_step(graphs[i], y[i], lr=0.05) return model # 4세대 모델 학습 (앱 시작 시 1번만) gnn_model = train_gnn() # ============================================================ # 0. 공통 설정 # ============================================================ FEATURES = ['금액', '시간', '신규수취인', '금액비율'] FEATURE_BG = { # 학습 데이터 평균값 (SHAP baseline) '금액': 180.0, '시간': 11.8, '신규수취인': 0.26, '금액비율': 2.8 } def build_training_data(): """학습 데이터 생성 (250건: 정상 200 + 사기 50)""" np.random.seed(42) normal = pd.DataFrame({ '금액': np.random.normal(100, 50, 200), '시간': np.random.normal(14, 4, 200), '신규수취인': np.random.binomial(1, 0.1, 200), '금액비율': np.random.normal(1, 0.5, 200), '라벨': 0 }) fraud = pd.DataFrame({ '금액': np.random.normal(500, 200, 50), '시간': np.random.normal(3, 2, 50), '신규수취인': np.random.binomial(1, 0.8, 50), '금액비율': np.random.normal(10, 5, 50), '라벨': 1 }) return pd.concat([normal, fraud], ignore_index=True) def train_gen2(): data = build_training_data() model = LogisticRegression(random_state=42, max_iter=1000) model.fit(data[FEATURES], data['라벨']) return model def train_gen3(): data = build_training_data() model = XGBClassifier(n_estimators=10, max_depth=3, learning_rate=0.1, random_state=42, eval_metric='logloss') model.fit(data[FEATURES], data['라벨']) return model gen2_model = train_gen2() gen3_model = train_gen3() # 학습된 파라미터 추출 (강의용 노출 목적) GEN2_COEF = gen2_model.coef_[0] GEN2_INTERCEPT = gen2_model.intercept_[0] GEN3_IMPORTANCE = gen3_model.feature_importances_ # ============================================================ # 0-C. XAI 헬퍼 함수 # ============================================================ # 3세대용: TreeSHAP을 직접 호출 (xgboost가 SHAP 값을 내부적으로 계산) # 4세대용: GNNExplainer 스타일의 엣지·노드 마스킹 기반 기여도 추출 def compute_shap_values_gen3(amount, hour, new_payee_bin, ratio): """XGBoost 내장 TreeSHAP으로 개별 거래의 SHAP 값 계산. pred_contribs=True 옵션 사용 시 [기여도_피처1, ..., 기여도_피처N, base_value] 반환. 합산하면 logit space에서의 모델 출력과 정확히 일치 (additive guarantee). """ import xgboost as xgb dmatrix = xgb.DMatrix( pd.DataFrame([[amount, hour, new_payee_bin, ratio]], columns=FEATURES) ) booster = gen3_model.get_booster() # pred_contribs=True → SHAP 값 직접 반환 shap_arr = booster.predict(dmatrix, pred_contribs=True)[0] # 마지막 원소는 base_value (= expected value over training data) base_value = float(shap_arr[-1]) feature_shap = [float(v) for v in shap_arr[:-1]] return feature_shap, base_value def compute_gnn_edge_attribution(amount, hour, new_payee_bin, ratio): """GNNExplainer 스타일: 각 엣지를 끄면 예측이 얼마나 떨어지는지 측정. 실제 GNNExplainer는 미분가능한 마스크를 학습하지만, 데모에서는 가장 직관적인 leave-one-edge-out 방식으로 단순화 (실무에서도 자주 쓰는 변형). sigmoid가 saturate되는 경우(prob≈1 또는 prob≈0)에는 확률 차이가 0에 가까워져 시각화가 안 되므로, logit-space에서 측정한 뒤 probability 차이도 함께 반환한다. """ node_features = build_graph_features(amount, hour, new_payee_bin, ratio) full_prob, full_inter = gnn_model.forward(node_features, return_intermediates=True) full_logit = full_inter['logit'] edge_info = [ (1, "송금인 → 거래"), (2, "수취인 → 거래"), (3, "단말기 → 거래"), ] contributions = [] for node_idx, label in edge_info: masked_adj = ADJ.astype(np.float32).copy() masked_adj[0, node_idx] = 0 masked_adj[node_idx, 0] = 0 row_sums = masked_adj.sum(axis=1, keepdims=True) row_sums[row_sums == 0] = 1 masked_adj_norm = masked_adj / row_sums masked_prob, masked_inter = gnn_model.forward( node_features, return_intermediates=True, adj_mask=masked_adj_norm ) masked_logit = masked_inter['logit'] # logit-space 차이 (saturate 영역에서도 유의미) logit_delta = full_logit - masked_logit # contributions 시그니처는 그대로 유지: (라벨, masked_prob, delta) # 단 delta는 logit 차이를 그대로 사용 → 시각화에서 의미가 살아남 contributions.append((label, masked_prob, logit_delta)) return full_prob, contributions def compute_gnn_node_feature_attribution(amount, hour, new_payee_bin, ratio): """GNNExplainer 스타일: 각 입력 피처를 baseline(평균값)으로 대체했을 때 예측이 얼마나 떨어지는지 측정 (피처 단위 기여도). 엣지 마스킹과 마찬가지로 logit-space에서 측정. """ full_node_features = build_graph_features(amount, hour, new_payee_bin, ratio) full_prob, full_inter = gnn_model.forward(full_node_features, return_intermediates=True) full_logit = full_inter['logit'] baselines = { '금액': 100.0, '시간': 14.0, '신규수취인': 0, '금액비율': 1.0 } inputs = { '금액': amount, '시간': hour, '신규수취인': new_payee_bin, '금액비율': ratio } contributions = [] for feat in FEATURES: masked_inputs = inputs.copy() masked_inputs[feat] = baselines[feat] masked_node_features = build_graph_features( masked_inputs['금액'], masked_inputs['시간'], masked_inputs['신규수취인'], masked_inputs['금액비율'] ) _, masked_inter = gnn_model.forward(masked_node_features, return_intermediates=True) masked_logit = masked_inter['logit'] logit_delta = full_logit - masked_logit contributions.append((feat, logit_delta)) return full_prob, contributions def build_counterfactual_gen5(amount, hour, new_payee_bin, ratio, prob_threshold=0.5): """5세대 보조: '이 거래가 통과되려면 무엇이 바뀌어야 했는가'를 탐색. GNN으로 후보를 빠르게 평가 (실제 LLM 호출 비용을 아끼기 위함). saturate 영역에서도 효과가 보이도록 logit space에서도 측정. """ candidates = [ ("금액을 100만원 이하로", lambda: build_graph_features(50, hour, new_payee_bin, ratio)), ("거래 시간을 14시(주간)로", lambda: build_graph_features(amount, 14, new_payee_bin, ratio)), ("기존 수취인이었다면", lambda: build_graph_features(amount, hour, 0, ratio)), ("평소 거래액 수준(1배)이었다면", lambda: build_graph_features(amount, hour, new_payee_bin, 1.0)), ] base_features = build_graph_features(amount, hour, new_payee_bin, ratio) base_prob, base_inter = gnn_model.forward(base_features, return_intermediates=True) base_logit = base_inter['logit'] results = [] for label, builder in candidates: cf_features = builder() cf_prob, cf_inter = gnn_model.forward(cf_features, return_intermediates=True) cf_logit = cf_inter['logit'] flipped = (base_prob >= prob_threshold and cf_prob < prob_threshold) # 시그니처: (라벨, cf_prob, prob_drop, flipped, logit_drop) results.append((label, cf_prob, base_prob - cf_prob, flipped, base_logit - cf_logit)) return results # ============================================================ # 1. 1세대 룰 정의 # ============================================================ GEN1_RULES = [ {"name": "R1 고액 거래", "condition": "금액 ≥ 500만원", "weight": 40}, {"name": "R2 새벽 시간대", "condition": "시간 ≤ 6 또는 ≥ 22", "weight": 30}, {"name": "R3 신규 수취인", "condition": "신규수취인 = 예", "weight": 20}, {"name": "R4 평소 대비 급증", "condition": "금액비율 ≥ 5배", "weight": 10}, ] def evaluate_gen1(amount, hour, new_payee_bin, ratio): triggered = [ amount >= 500, hour >= 22 or hour <= 6, new_payee_bin == 1, ratio >= 5, ] score = sum(r["weight"] for r, t in zip(GEN1_RULES, triggered) if t) return triggered, score def decide(prob_or_score, is_score=False): if is_score: if prob_or_score >= 70: return "차단", "#FCEBEB", "#791F1F" if prob_or_score >= 40: return "추가 인증", "#FAEEDA", "#854F0B" return "통과", "#EAF3DE", "#3B6D11" else: if prob_or_score >= 0.7: return "차단", "#FCEBEB", "#791F1F" if prob_or_score >= 0.5: return "추가 인증", "#FAEEDA", "#854F0B" return "통과", "#EAF3DE", "#3B6D11" # ============================================================ # 2. 공통 HTML 빌더 # ============================================================ def card_header(gen_label, title, decision_text, bg_color, text_color, sub): return f"""

{gen_label}

{title}

{decision_text}

{sub}

""" def formula_box(html): return f"""
{html}
""" def feature_setup_box(actor_label, actor_color, items, explanation): color_map = { 'human': ('#E6F1FB', '#0C447C'), 'model': ('#FAECE7', '#993C1D'), 'mixed': ('#F1EFE8', '#5F5E5A'), } badge_bg, badge_fg = color_map.get(actor_color, color_map['mixed']) rows = "" for name, actor, desc in items: a_bg, a_fg = color_map.get(actor, color_map['mixed']) rows += ( f"" f"{name}" f"" f"{actor}" f"" f"{desc}" f"" ) return f"""

⚙️ Feature·Rule 결정 방식

{actor_label}
{rows}

{explanation}

""" def xai_box(method, status, era, items, takeaway): """세대별 XAI 구현 방식을 표시하는 박스 (새로 추가). method: XAI 기법 명칭 (예: 'TreeSHAP', 'GNNExplainer') status: 'native' (내재) | 'post-hoc' (사후) | 'none' (불필요) | 'generative' (생성형) era: 해당 기법이 표준화된 시기 items: [(항목, 설명)] 리스트 takeaway: 강의 포인트 한 줄 """ status_map = { 'none': ('XAI 불필요', '#EAF3DE', '#3B6D11'), 'native': ('내재적 설명력', '#E6F1FB', '#0C447C'), 'post-hoc': ('사후 설명 기법', '#FAEEDA', '#854F0B'), 'generative': ('생성형 설명', '#FAECE7', '#993C1D'), } status_label, badge_bg, badge_fg = status_map.get(status, status_map['post-hoc']) rows = "" for name, desc in items: rows += ( f"" f"{name}" f"{desc}" f"" ) return f"""

🔍 XAI 구현 방식

{status_label} 기법: {method} · {era}
{rows}

💡 {takeaway}

""" CARD_STYLE = ("background:#fff; border:0.5px solid rgba(0,0,0,0.15); " "border-radius:12px; padding:16px 20px; margin-bottom:14px;") # ============================================================ # 3. 세대별 HTML 생성 함수 # ============================================================ def render_gen1(amount, hour, new_payee_bin, ratio): triggered, score = evaluate_gen1(amount, hour, new_payee_bin, ratio) dec, bg, fg = decide(score, is_score=True) rows = "" for rule, t in zip(GEN1_RULES, triggered): applied = rule["weight"] if t else 0 row_bg = "#FAECE7" if t else "#ffffff" td_color = "#4A1B0C" if t else "#444" sub_color = "#712B13" if t else "#666" mark = "✓" if t else "—" rows += f""" {rule['name']} {rule['condition']} +{rule['weight']} {mark} +{applied} """ gen1_setup = feature_setup_box( actor_label="100% 사람 결정", actor_color='human', items=[ ("입력 Feature 4개", "사람", "도메인 전문가가 '금액·시간·신규수취인·금액비율'을 사기 판단 기준으로 선정"), ("룰 조건 (임계값)", "사람", "≥500만원, ≤6시 또는 ≥22시, =1, ≥5배 — 모두 사람이 직접 결정"), ("룰별 가중치", "사람", "40 / 30 / 20 / 10점 — 도메인 경험에 따라 사람이 부여"), ("판정 임계값", "사람", "70점 이상 차단, 40점 이상 추가인증 — 운영팀이 비즈니스 판단으로 결정"), ], explanation="모든 결정이 사람의 도메인 지식에 의존. 학습 데이터는 사용하지 않음. 새 사기 패턴 등장 시 사람이 룰을 추가해야 함." ) # ─── XAI 구현 박스 (1세대) ───────────────────────────── gen1_xai = xai_box( method="해당 없음 (Self-Explanatory)", status='none', era="~ 2000년대 중반", items=[ ("설명 생성 방식", "룰북 자체가 설명서. 발동된 룰의 이름·조건·가중치를 그대로 노출하면 사람이 즉시 이해 가능"), ("개별 거래 설명", "위 표가 바로 설명. '왜 차단됐냐 → R1(고액)+R3(신규수취인)+R4(급증) = 70점'으로 즉시 답변"), ("감독당국 대응", "룰 매트릭스 그대로 제출. 별도 XAI 알고리즘 필요 없음"), ("한계", "룰이 많아지면(수백 개) 사람도 따라가기 힘들어짐 → 룰 간 상호작용·우선순위가 새로운 블랙박스가 됨"), ], takeaway="1세대에는 '설명력'이라는 개념이 따로 존재하지 않았음. 판단 로직 = 설명 로직이기 때문. XAI라는 단어가 등장한 것은 모델이 비선형으로 진화한 후의 일." ) return f"""
{card_header("GEN 1 · RULE-BASED", "규칙 기반 판단", dec, bg, fg, f"누적 {score}점 / 100점")} {gen1_setup} {gen1_xai} {formula_box("총점 = Σ (발동된 룰의 가중치) → 임계값 비교 (≥70 차단 / ≥40 추가인증)")} {rows}
룰 조건 가중치 발동 적용
최종 합계 {score}점

한계: 룰이 고정값이라 임계값 바로 아래(예: 499만원 23시) 거래를 놓침

""" def render_gen2(amount, hour, new_payee_bin, ratio): input_vec = np.array([amount, hour, new_payee_bin, ratio], dtype=float) contributions = GEN2_COEF * input_vec logit = contributions.sum() + GEN2_INTERCEPT prob = 1 / (1 + np.exp(-logit)) dec, bg, fg = decide(prob) rows = "" for f, x, w, c in zip(FEATURES, input_vec, GEN2_COEF, contributions): if c > 0: row_bg, td_c, sub_c = "#FAECE7", "#4A1B0C", "#712B13" elif c < 0: row_bg, td_c, sub_c = "#E1F5EE", "#04342C", "#085041" else: row_bg, td_c, sub_c = "#ffffff", "#444", "#666" rows += f""" {f} {x:.3f} {w:+.4f} {c:+.4f} """ contrib_str = " + ".join([f"({c:+.4f})" for c in contributions]) calc_html = ( f"z = {contrib_str} + ({GEN2_INTERCEPT:+.4f})
" f"z = {logit:+.4f}
" f"P = 1 / (1 + e{-logit:+.4f}) = " f"{prob:.4f} → {prob*100:.2f}%" ) gen2_setup = feature_setup_box( actor_label="피처는 사람, 가중치는 모델", actor_color='mixed', items=[ ("입력 Feature 4개", "사람", "1세대와 동일한 4개 컬럼을 사람이 선정 (피처 엔지니어링)"), ("학습 데이터", "사람", "250건의 거래에 사기/정상 라벨을 사람이 부여"), ("가중치 w₁~w₄", "모델", "fit() 호출 시 L-BFGS 알고리즘이 자동 학습"), ("절편 b", "모델", "데이터의 사기 비율(50/250=20%)에 맞춰 자동 조정"), ("판정 임계값", "사람", "0.5(추가인증) / 0.7(차단) — 운영팀이 결정"), ], explanation="피처는 여전히 사람이 정의. 모델이 학습하는 건 '4개 피처에 어떤 가중치를 곱해야 사기를 잘 맞추는가'뿐." ) # ─── XAI 구현 박스 (2세대) ───────────────────────────── # 최대 기여 피처 찾기 top_idx = int(np.argmax(np.abs(contributions))) top_feat = FEATURES[top_idx] top_contrib = contributions[top_idx] gen2_xai = xai_box( method="Coefficient-based Attribution (계수 분해)", status='native', era="통계 모델 시대부터 자연 발생", items=[ ("설명 생성 방식", "수식 z = Σ(wᵢ·xᵢ) + b 가 그대로 설명. 별도 XAI 알고리즘 없이 가중치만 보면 됨"), ("개별 거래 설명", f"본 거래에서 가장 큰 기여 = {top_feat} ({top_contrib:+.4f}). 표의 마지막 컬럼이 곧 SHAP 값의 정확한 선형 버전"), ("전역 설명 (global)", "가중치 부호와 크기가 곧 변수 중요도. 양수 = 사기 방향, 음수 = 정상 방향"), ("한계", "선형 가정이라 비선형 패턴 학습 불가. '금액 + 시간 조합' 같은 상호작용을 못 잡음 → 정확도 부족이 3세대 등장의 배경"), ], takeaway="로지스틱 회귀의 'wᵢ·xᵢ' 분해는 사실상 정확한 SHAP 값과 동치(선형 모델 한정). XAI라는 개념이 별도로 필요 없는 마지막 세대." ) return f"""
{card_header("GEN 2 · LOGISTIC REGRESSION", "로지스틱 회귀 (선형 모델)", dec, bg, fg, f"사기 확률 {prob*100:.2f}%")} {gen2_setup} {gen2_xai} {formula_box("z = w₁·금액 + w₂·시간 + w₃·신규수취인 + w₄·금액비율 + b
P(사기) = 1 / (1 + e-z)")} {rows}
피처 입력값 x 학습 가중치 w 기여도 w·x
절편 (bias) b {GEN2_INTERCEPT:+.4f}
{formula_box(calc_html)}

해석: 가중치 부호가 곧 판단 방향. 양수는 사기 쪽, 음수는 정상 쪽으로 끌어당김

""" def render_gen3(amount, hour, new_payee_bin, ratio): input_df = pd.DataFrame([[amount, hour, new_payee_bin, ratio]], columns=FEATURES) prob = float(gen3_model.predict_proba(input_df)[0][1]) dec, bg, fg = decide(prob) imp_pairs = sorted(zip(FEATURES, GEN3_IMPORTANCE), key=lambda x: -x[1]) max_imp = max(GEN3_IMPORTANCE) if max(GEN3_IMPORTANCE) > 0 else 1 imp_bars = "" for f, imp in imp_pairs: bar_w = (imp / max_imp) * 100 imp_bars += f"""
{f}
{imp:.3f}
""" # ──────────────────────────────────────────────────────────── # 진짜 학습된 트리 10개 전체에서 본 거래가 도달한 leaf 값을 추출 # ──────────────────────────────────────────────────────────── booster = gen3_model.get_booster() trees_df = booster.trees_to_dataframe() input_values = {'금액': amount, '시간': hour, '신규수취인': new_payee_bin, '금액비율': ratio} tree_traces = [] for tree_id in range(10): tree = trees_df[trees_df['Tree'] == tree_id].set_index('ID') current_id = f"{tree_id}-0" path = [] leaf_val = 0.0 while True: row = tree.loc[current_id] if row['Feature'] == 'Leaf': leaf_val = float(row['Gain']) break feature = row['Feature'] split = float(row['Split']) input_v = input_values[feature] if input_v < split: path.append(f"[{feature} < {split:.2f}] Yes") current_id = row['Yes'] else: path.append(f"[{feature} < {split:.2f}] No") current_id = row['No'] tree_traces.append((path, leaf_val)) raw_score = sum(leaf for _, leaf in tree_traces) tree_rows = "" cumulative = 0.0 for tree_id, (path, leaf) in enumerate(tree_traces): cumulative += leaf path_text = " → ".join(path) + f" → leaf={leaf:+.4f}" leaf_color = "#4A1B0C" if leaf > 0 else "#04342C" leaf_bg = "#FAECE7" if leaf > 0 else "#E1F5EE" tree_rows += ( f"" f"#{tree_id}" f"{path_text}" f"{leaf:+.4f}" f"{cumulative:+.4f}" f"" ) sigmoid_result = 1 / (1 + np.exp(-raw_score)) tree_table = f""" {tree_rows}
트리 본 거래의 분기 경로 → 도달한 leaf leaf 값 누적 raw score
최종 raw score (10개 트리 합산) {raw_score:+.4f}
""" final_calc = formula_box( f"P(사기) = sigmoid({raw_score:+.4f}) " f"= 1 / (1 + e{-raw_score:+.4f}) " f"= {sigmoid_result:.4f} ({sigmoid_result*100:.2f}%)" ) # ─── TreeSHAP 계산 및 시각화 ────────────────────────── shap_values, shap_base = compute_shap_values_gen3(amount, hour, new_payee_bin, ratio) # SHAP 검증: base + Σ(shap) ≈ raw_score (logit space) shap_sum = shap_base + sum(shap_values) # SHAP waterfall: 절댓값 큰 순으로 정렬 shap_pairs = sorted(zip(FEATURES, shap_values), key=lambda x: -abs(x[1])) max_abs_shap = max(abs(v) for v in shap_values) if any(shap_values) else 1.0 shap_rows = "" for feat, sv in shap_pairs: direction = "사기 ↑" if sv > 0 else "정상 ↓" color = "#A32D2D" if sv > 0 else "#3B6D11" bar_bg = "#FAECE7" if sv > 0 else "#E1F5EE" bar_w = (abs(sv) / max_abs_shap) * 100 if max_abs_shap > 0 else 0 bar_align = "flex-start" if sv > 0 else "flex-end" # 좌우로 분리된 막대 (양수는 오른쪽, 음수는 왼쪽) if sv > 0: bar_html = f"""
""" else: bar_html = f"""
""" shap_rows += f""" {feat} {sv:+.4f} {bar_html} {direction} """ shap_table = f""" {shap_rows}
피처 SHAP 값 ← 정상 방향 | 사기 방향 → 기여
base value (평균 거래) {shap_base:+.4f} 학습 데이터 전체의 평균 logit
합계 = base + Σ(SHAP) {shap_sum:+.4f} ≈ raw score {raw_score:+.4f} (가산성 보장)
""" gen3_setup = feature_setup_box( actor_label="피처는 사람, 트리 구조는 모델", actor_color='mixed', items=[ ("입력 Feature 4개", "사람", "1·2세대와 완전히 동일한 4개 컬럼"), ("학습 데이터", "사람", "1·2세대와 동일한 250건"), ("트리 분기 임계값", "모델", "Gain 최대화로 자동 결정 (예: 시간 < 8.69, 금액비율 < 1.77)"), ("각 leaf 값", "모델", "각 leaf에 도달한 샘플들의 잔차로 자동 계산"), ("트리 개수 / 깊이", "사람", "n_estimators=10, max_depth=3 (하이퍼파라미터)"), ("판정 임계값", "사람", "0.5(추가인증) / 0.7(차단)"), ], explanation="2세대보다 학습되는 부분이 훨씬 많아짐. 분기 임계값과 leaf 값 모두 데이터에서 자동 발견." ) # ─── XAI 구현 박스 (3세대) ───────────────────────────── top_shap_feat, top_shap_val = shap_pairs[0] gen3_xai = xai_box( method="TreeSHAP (Lundberg 2017)", status='post-hoc', era="2017년 발표 → 2018-2020년 금융권 표준화", items=[ ("설명 생성 방식", "모델 학습 후 별도 알고리즘으로 SHAP 값 계산. XGBoost는 TreeSHAP이라는 트리 구조 특화 알고리즘이 내장되어 있어 실시간 추론에도 사용 가능 (개별 거래당 ~1ms)"), ("개별 거래 설명", f"본 거래의 SHAP 분해: 가장 큰 기여는 {top_shap_feat} ({top_shap_val:+.4f}). 부호가 양수면 사기 방향, 음수면 정상 방향으로 끌어당김"), ("수학적 보장", f"가산성(Additive): base({shap_base:+.4f}) + Σ(SHAP) = {shap_sum:+.4f} ≈ raw score {raw_score:+.4f}. 즉 '평균 거래 대비 이 거래가 왜 더 의심스러운가'를 정확히 분해"), ("금융권 활용", "감독당국 보고서(설명가능성 자료), 고객 거절 사유 통보(Adverse Action Notice), 모델 디버깅·검증"), ("한계", "어디까지나 '근사된 기여도'. 모델이 실제로 그렇게 사고했다는 보장은 아님 → 4세대 GNN에서는 더 큰 한계가 됨"), ], takeaway="3세대부터 'XAI'가 본격적으로 별도 모듈로 등장. 모델 = 판단기, SHAP = 설명기로 역할이 분리됨. 이게 현재 금융권 FDS의 표준 아키텍처." ) return f"""
{card_header("GEN 3 · XGBOOST (TREE ENSEMBLE)", "XGBoost (트리 10개 앙상블)", dec, bg, fg, f"사기 확률 {prob*100:.2f}%")} {gen3_setup} {gen3_xai} {formula_box("F(x) = Σk=1..K fk(x),   fk ∈ 트리 공간
P(사기) = sigmoid(F(x))  [K=10, max_depth=3, lr=0.1]")}

피처 중요도 (Gain 기반, 전역 설명)

{imp_bars}

⚠️ 위 중요도는 모델 전체 평균이라 개별 거래 설명에는 부적합 → 그래서 SHAP이 필요

🎯 TreeSHAP — 본 거래에 대한 개별 기여도 분해 (Local Explanation)

'이 거래가 평균보다 왜 더 의심스러운가'를 피처별로 정량 분해. 합산하면 모델의 raw score와 일치 (가산성 보장).

{shap_table}

🌳 학습된 트리 10개의 leaf 값 누적 (부스팅 본질)

각 트리가 이전 트리의 잔차를 보정하며 leaf 값을 더해감 → 누적된 raw score를 sigmoid로 변환

{tree_table}

🧮 최종 확률 계산

{final_calc}

강점: 비선형 패턴·피처 상호작용 자동 학습 + SHAP으로 개별 설명 확보. 한계: 학습 데이터 분포 밖의 케이스(예: 전세 잔금)는 여전히 못 잡음

""" def render_gen4(amount, hour, new_payee_bin, ratio, prob3): """4세대 GNN - 진짜 학습된 mini GNN의 forward pass 결과 + GNNExplainer 스타일 XAI""" node_features = build_graph_features(amount, hour, new_payee_bin, ratio) prob, intermediates = gnn_model.forward(node_features, return_intermediates=True) prob = float(prob) dec, bg, fg = decide(prob) h1 = intermediates['h1'] h2 = intermediates['h2'] trans_emb = intermediates['trans_embedding'] logit = intermediates['logit'] # 그래프 시각화 색상 edge2_color = "#D85A30" if new_payee_bin == 1 else "#888780" edge2_dash = 'stroke-dasharray="" ' if new_payee_bin == 1 else 'stroke-dasharray="3,3" ' risk_fill = "#F7C1C1" if new_payee_bin == 1 else "#D3D1C7" risk_stroke = "#A32D2D" if new_payee_bin == 1 else "#5F5E5A" risk_text = "사기계좌" if new_payee_bin == 1 else "일반" risk_color = "#501313" if new_payee_bin == 1 else "#444441" payee_type = "신규" if new_payee_bin == 1 else "기존" svg = f""" 송금인 정상이력 95% 단말기 신규 IP 본 거래 {amount:.0f}만 / {int(hour):02d}시 수취인 {payee_type} 1-hop {risk_text} 일반 1-hop 이웃 2-hop 이웃 """ # 16차원 임베딩을 4×4 그리드로 시각화 + 색상 강조 h1_trans = h1[0] h2_trans = h2[0] def render_node_grid(values, prefix, base_x, base_y): cells = "" max_v = max(abs(values.max()), abs(values.min())) if len(values) > 0 else 1.0 for i, v in enumerate(values): r, c = i // 4, i % 4 x = base_x + c * 16 y = base_y + r * 16 intensity = abs(v) / max_v if max_v > 0 else 0 if v > 0: fill = f"rgb({int(240 - 100*intensity)}, {int(160 - 70*intensity)}, {int(150 - 70*intensity)})" stroke = "#993C1D" text_color = "#4A1B0C" if intensity > 0.4 else "#5F4308" else: fill = f"rgb({int(220 - 60*intensity)}, {int(220 - 30*intensity)}, {int(210 - 50*intensity)})" stroke = "#888" text_color = "#555" cells += ( f'' f'{v:.1f}' ) return cells grid_h1 = render_node_grid(h1_trans, "h", 195, 55) grid_h2 = render_node_grid(h2_trans, "h'", 395, 55) feature_expansion_svg = f""" 입력층 사람이 정의 은닉층 1 (1-hop) 모델이 생성 은닉층 2 (2-hop) 모델이 생성 출력 사람이 정의 금액 = {amount:.0f} 시간 = {int(hour)} 신규수취인 = {new_payee_bin} 금액비율 = {ratio:.1f} 4 features 의미: 명확 {grid_h1} 16 features 의미: 모름 {grid_h2} 16 features 의미: 모름 사기 확률 {prob:.4f} W₁ 4×16=64 W₂ 16×16=256 W₃ (MLP) 16→1 총 학습 파라미터: 337개 (W₁:64 + W₂:256 + W_mlp:16 + bias:1) 학습 데이터 250건 · 100 epoch · 학습 시간 약 2초 (앱 시작 시 1번) vs 2세대 로지스틱 회귀 5개 파라미터 → 약 67배 증가, 표현력↑ 해석가능성↓ """ # Layer별 활성도 layer1_active = (h1_trans > 0).sum() layer2_active = (h2_trans > 0).sum() layer1_mean = float(h1_trans[h1_trans > 0].mean()) if layer1_active > 0 else 0.0 layer2_mean = float(h2_trans[h2_trans > 0].mean()) if layer2_active > 0 else 0.0 # ─── GNNExplainer 스타일 XAI: 엣지 마스킹 (logit space) ─ # NOTE: delta는 logit 차이. sigmoid가 saturate(prob≈0 또는 1)되어도 # logit space에서는 변화가 그대로 보존되므로 시각화에 적합. _, edge_contribs = compute_gnn_edge_attribution(amount, hour, new_payee_bin, ratio) max_edge_delta = max(abs(d) for _, _, d in edge_contribs) if edge_contribs else 1.0 edge_rows = "" for label, masked_p, delta in edge_contribs: # logit space 기준 임계값 (0.1 이상이면 유의미) if abs(delta) < 0.1: interp = "거의 영향 없음" color, row_bg = "#888", "#ffffff" elif delta > 0: interp = "이 엣지를 끄면 위험도 ↓ → 이 엣지가 위험을 만드는 핵심" color, row_bg = "#A32D2D", "#FAECE7" else: interp = "이 엣지를 끄면 위험도 ↑ → 이 엣지가 안전 신호였음" color, row_bg = "#3B6D11", "#E1F5EE" bar_w = (abs(delta) / max_edge_delta) * 100 if max_edge_delta > 0 else 0 edge_rows += f""" {label} {masked_p:.4f} {delta:+.4f}
{interp} """ edge_table = f""" {edge_rows}
제거 대상 엣지 제거 후 P Δ logit 크기 해석
기준선 (전체 그래프) {prob:.4f} 엣지를 모두 살린 원본 예측. ※ P는 sigmoid 후 값이라 saturate 가능 → 기여도는 logit space에서 측정
""" # ─── GNNExplainer 스타일 XAI: 노드 피처 마스킹 ───────── _, node_feat_contribs = compute_gnn_node_feature_attribution(amount, hour, new_payee_bin, ratio) max_feat_delta = max(abs(d) for _, d in node_feat_contribs) if node_feat_contribs else 1.0 feat_rows = "" for feat, delta in sorted(node_feat_contribs, key=lambda x: -abs(x[1])): color = "#A32D2D" if delta > 0 else ("#3B6D11" if delta < 0 else "#888") row_bg = "#FAECE7" if delta > 0 else ("#E1F5EE" if delta < 0 else "#ffffff") bar_w = (abs(delta) / max_feat_delta) * 100 if max_feat_delta > 0 else 0 feat_rows += f""" {feat} {delta:+.4f}
""" feat_table = f""" {feat_rows}
피처 (baseline으로 대체 시) Δ logit 크기
""" # 가장 강하게 활성화된 차원 top 3 top3_indices = np.argsort(h2_trans)[-3:][::-1] top3_rows = "" for i, idx in enumerate(top3_indices): val = h2_trans[idx] if val < 0.01: interpretation = "(거의 활성화 안 됨)" color = "#888" bg_row = "#ffffff" else: interpretation_pool = [ "수취인 위험도 신호 추정 (모델만 아는 추상 패턴)", "송금인 평소 행동 이탈도 추정", "거래 시간대 + 금액 조합 신호", "사람이 해석 불가 (모델 내부 표현)", "그래프 2-hop 위험 클러스터 신호 추정", ] interpretation = interpretation_pool[i % len(interpretation_pool)] color = "#4A1B0C" if val > 0.5 else "#5F4308" bg_row = "#FAECE7" if val > 0.5 else "#FAEEDA" top3_rows += ( f"" f"h'{idx} (은닉층 2)" f"{val:+.3f}" f"{interpretation}" f"" ) interpret_table = f""" {top3_rows}
차원 활성값 사람의 추정 (모델은 알려주지 않음)
나머지 13개 차원: 대부분 해석 불가
""" gen4_setup = feature_setup_box( actor_label="구조는 사람, 임베딩은 모델 생성", actor_color='model', items=[ ("그래프 구조 정의", "사람", "노드 종류(거래·송금인·수취인·단말기)와 엣지 관계는 사람이 설계"), ("학습 데이터", "사람", "1-3세대와 동일한 250건 거래에 라벨 부여"), ("노드 초기 임베딩", "사람", "각 노드의 4차원 초기값은 사람이 인코딩 규칙 작성"), ("은닉층 16차원 Feature", "모델", "사람이 정의 안 함. 모델이 학습으로 16개 익명 차원을 자동 생성"), ("가중치 W₁, W₂, W_mlp", "모델", "337개 파라미터를 backpropagation으로 자동 학습"), ("판정 임계값", "사람", "0.5(추가인증) / 0.7(차단)"), ], explanation="결정적 차이: 1-3세대는 사람이 정한 4개 피처만 봤지만, 4세대는 모델이 16+16=32개의 새 익명 피처를 스스로 만들어냄." ) # ─── XAI 구현 박스 (4세대) ───────────────────────────── # 가장 영향력 큰 엣지·피처 찾기 top_edge_label, _, top_edge_delta = max(edge_contribs, key=lambda x: abs(x[2])) top_feat_name, top_feat_delta = max(node_feat_contribs, key=lambda x: abs(x[1])) gen4_xai = xai_box( method="GNNExplainer (Ying 2019) — 엣지·노드 마스킹", status='post-hoc', era="2019년 발표 → 2021-2023년 그래프 FDS 도입", items=[ ("왜 SHAP만으론 부족한가", "GNN은 입력이 '거래 1건의 피처'가 아니라 '거래를 둘러싼 그래프(노드+엣지) 전체'. SHAP은 피처 기여도만 분해할 뿐, '어느 이웃과의 관계가 위험을 만들었는가'는 답하지 못함"), ("엣지 마스킹 (Edge Attribution)", f"각 엣지를 차례로 끄고 예측 변화 측정 (logit space). 본 거래에서 가장 큰 영향 엣지 = {top_edge_label} (Δ logit = {top_edge_delta:+.4f}). 이게 그래프 구조 설명의 핵심"), ("노드 피처 마스킹", f"각 입력 피처를 평균값으로 대체하고 예측 변화 측정. 본 거래에서 가장 큰 영향 피처 = {top_feat_name} (Δ logit = {top_feat_delta:+.4f})"), ("logit space 측정 이유", "사기 케이스에서 sigmoid가 saturate(P≈1.0)되면 확률 차이가 0에 수렴해 시각화 불가. 그래서 sigmoid 직전의 logit에서 차이를 측정 (실무 GNNExplainer 구현도 동일)"), ("실무 운영 가치", "예: '이 거래가 차단된 이유: 수취인이 사기계좌와 2-hop 거리에 있기 때문' 같은 그래프 기반 설명이 가능 → 콜센터·심사팀이 고객 응대에 활용"), ("한계", "여전히 16차원 은닉 임베딩 자체의 의미는 해석 불가. 위 '활성화 상위 3차원'에 적힌 해석은 모두 사람의 사후 추측일 뿐"), ], takeaway="4세대 XAI의 핵심은 '피처 기여도 → 그래프 구조 기여도'로 설명 단위가 확장된 것. 보이스피싱 클러스터 탐지 같은 작업에서 결정적으로 유용함." ) return prob, f"""
{card_header("GEN 4 · GNN (GRAPH NEURAL NETWORK)", "그래프 신경망 (numpy 학습)", dec, bg, fg, f"사기 확률 {prob*100:.2f}%")} {gen4_setup} {gen4_xai} {formula_box("hv(l+1) = ReLU(W(l) · AGG({{hu(l) : u ∈ N(v)}}))
P(사기) = sigmoid(Wmlp · h거래(2))   [2-hop 메시지 패싱, 250건 학습됨]")}

2-hop 이웃 그래프

{svg}

🎯 GNNExplainer — 엣지 기여도 (Edge Attribution)

"이 거래가 위험한 이유가 어떤 이웃과의 관계 때문인가?" → 엣지를 하나씩 끄면서 예측이 얼마나 떨어지는지 측정 (leave-one-edge-out)

{edge_table}

🎯 GNNExplainer — 노드 피처 기여도 (Node Feature Attribution)

"이 거래의 어떤 속성이 가장 위험을 만들었나?" → 각 피처를 평균값(baseline)으로 대체하고 예측 변화 측정

{feat_table}

🔍 Feature가 어떻게 확장되는가 (실제 학습된 가중치로 forward pass)

사람이 정한 4개 → 모델이 만든 16개 → 또 다른 16개 → 사기 확률

{feature_expansion_svg}

📊 Layer별 활성화 통계

계층 집계 내용 활성 차원 평균 활성값
Layer 1 (1-hop) 송금인+수취인+단말기 임베딩 집계 {layer1_active}/16 {layer1_mean:.3f}
Layer 2 (2-hop) Layer 1 결과를 다시 한 hop 전파 {layer2_active}/16 {layer2_mean:.3f}
MLP (분류기) 거래 노드 임베딩 → logit → sigmoid logit={logit:+.2f} P={prob:.3f}

🔬 가장 강하게 활성화된 차원 (사후 추정 — 모델 내부는 알 수 없음)

{interpret_table}

강점: 그래프 구조(관계망)를 학습 + GNNExplainer로 '어느 이웃·어느 피처가 결정에 기여했는지' 추출. 한계: 16차원 임베딩 자체의 의미는 여전히 블랙박스

""" # ============================================================ # 5세대 — Claude API 실제 호출 # ============================================================ def build_claude_prompt(amount, hour, new_payee_bin, ratio, prior_avg, prior_dec, gen3_shap=None, gen4_edges=None, gen4_feats=None): """Claude에게 보낼 시스템 프롬프트와 사용자 메시지 구성. 3·4세대의 XAI 결과(SHAP, GNNExplainer)를 컨텍스트로 함께 주입하여 LLM이 단순한 자체 추론이 아니라 하위 모델의 설명까지 참조하도록 함. 이게 실무에서 권장되는 'Grounded Reasoning' 패턴. """ system_prompt = ( "당신은 한국 은행의 FDS(이상금융거래탐지시스템) 분석 전문가입니다. " "주어진 거래 정보와 1-4세대 모델의 사전 판단·XAI 분석 결과를 바탕으로 사기 여부를 종합 판정하세요. " "특히 부동산 잔금 송금, 사업자 대금 결제 같은 정상 거래 패턴과 보이스피싱·대포통장 패턴을 구분해야 합니다.\n\n" "분석 결과는 반드시 submit_fds_judgment 도구를 사용해서 제출하세요. " "reasoning_steps는 3~5개로, 각 단계마다 어떤 신호를 어떻게 해석했는지 한 줄로 쓰고 " "attention 가중치(0.0~1.0)를 부여하세요. attention의 합은 1.0 근처가 되도록 분배하세요. " "judgment 필드에는 판단 사유와 권고 조치를 함께 자연어로 작성하세요. " "counterfactual 필드에는 '이 거래가 통과되려면 무엇이 달라져야 했는가'를 1~2문장으로 적으세요." ) payee_str = "신규" if new_payee_bin == 1 else "기존" time_period = "새벽" if (hour <= 6 or hour >= 22) else ("주간" if 9 <= hour <= 18 else "저녁") # 3·4세대 XAI 결과를 텍스트로 정리 xai_context = "" if gen3_shap is not None: shap_lines = ", ".join([f"{f}={v:+.3f}" for f, v in gen3_shap]) xai_context += f"\n[3세대 XGBoost SHAP 분해] {shap_lines}\n" if gen4_edges is not None: edge_lines = ", ".join([f"{lbl}={d:+.3f}" for lbl, _, d in gen4_edges]) xai_context += f"[4세대 GNN 엣지 기여도] {edge_lines}\n" if gen4_feats is not None: feat_lines = ", ".join([f"{f}={d:+.3f}" for f, d in gen4_feats]) xai_context += f"[4세대 GNN 노드피처 기여도] {feat_lines}\n" user_message = ( f"[거래 정보]\n" f"- 금액: {amount:.0f}만원\n" f"- 거래 시간: {int(hour):02d}시 ({time_period})\n" f"- 수취인: {payee_str} 수취인\n" f"- 과거 대비 배율: {ratio:.1f}배 (송금인의 평균 거래액 대비)\n\n" f"[1-4세대 모델 사전 판단]\n" f"- 평균 사기 확률: {prior_avg*100:.1f}%\n" f"- 종합 판정: {prior_dec}\n" f"{xai_context}\n" f"위 거래에 대해 1-4세대 XAI 결과를 참조하여 FDS 전문가 관점에서 종합 판정해주세요. " f"특히 XAI 결과 중 가장 중요한 신호가 무엇이었는지 reasoning에 반영하세요." ) return system_prompt, user_message # Tool Use 스키마 정의 — counterfactual 필드 추가 FDS_JUDGMENT_TOOL = { "name": "submit_fds_judgment", "description": "FDS 분석 결과를 구조화된 형식으로 제출합니다.", "input_schema": { "type": "object", "properties": { "risk_score": { "type": "number", "description": "사기 의심도 (0.0~1.0)", "minimum": 0.0, "maximum": 1.0 }, "decision": { "type": "string", "description": "최종 판정", "enum": ["차단", "추가 인증", "통과"] }, "reasoning_steps": { "type": "array", "description": "추론 단계 3~5개", "minItems": 3, "maxItems": 5, "items": { "type": "object", "properties": { "step": {"type": "string", "description": "추론 내용 (한 줄)"}, "attention": {"type": "number", "minimum": 0.0, "maximum": 1.0}, "evidence_source": { "type": "string", "description": "이 추론 단계의 근거 출처 (예: '3세대 SHAP', '4세대 GNNExplainer', '시간 정보', '도메인 지식')", } }, "required": ["step", "attention"] } }, "judgment": { "type": "string", "description": "최종 자연어 판단 (왜 그렇게 판단했는지 + 권고 조치를 한국어로)" }, "counterfactual": { "type": "string", "description": "이 거래가 통과되려면 무엇이 달라져야 했는가 (반사실 설명)" } }, "required": ["risk_score", "decision", "reasoning_steps", "judgment"] } } def _try_parse_json_with_repair(text): try: return json.loads(text) except json.JSONDecodeError: pass repaired = re.sub(r',(\s*[}\]])', r'\1', text) try: return json.loads(repaired) except json.JSONDecodeError: pass repaired2 = re.sub(r'[\x00-\x1f]', lambda m: f'\\u{ord(m.group()):04x}', repaired) return json.loads(repaired2) def call_claude_api(amount, hour, new_payee_bin, ratio, prior_avg, prior_dec, gen3_shap=None, gen4_edges=None, gen4_feats=None): if not CLAUDE_AVAILABLE: return None, "API 키 미설정 (ANTHROPIC_API_KEY 환경변수 없음)" system_prompt, user_message = build_claude_prompt( amount, hour, new_payee_bin, ratio, prior_avg, prior_dec, gen3_shap=gen3_shap, gen4_edges=gen4_edges, gen4_feats=gen4_feats ) try: t0 = time.time() response = claude_client.messages.create( model=CLAUDE_MODEL, max_tokens=1024, temperature=0.2, system=system_prompt, tools=[FDS_JUDGMENT_TOOL], tool_choice={"type": "tool", "name": "submit_fds_judgment"}, messages=[{"role": "user", "content": user_message}] ) latency = time.time() - t0 parsed = None raw_text = "" for block in response.content: if block.type == "tool_use" and block.name == "submit_fds_judgment": parsed = block.input raw_text = json.dumps(parsed, ensure_ascii=False, indent=2) break elif block.type == "text": raw_text += block.text if parsed is None: json_match = re.search(r'\{.*\}', raw_text, re.DOTALL) if not json_match: return None, f"응답에 JSON 없음. raw_text 앞부분: {raw_text[:200]}" try: parsed = _try_parse_json_with_repair(json_match.group(0)) except json.JSONDecodeError as e: return None, f"JSONDecodeError: {str(e)[:150]} | raw 앞부분: {raw_text[:200]}" meta = { "input_tokens": response.usage.input_tokens, "output_tokens": response.usage.output_tokens, "latency": latency, "model": CLAUDE_MODEL, "raw_text": raw_text, } return parsed, meta except Exception as e: return None, f"API 호출 실패: {type(e).__name__}: {str(e)[:200]}" def render_gen5_card(amount, hour, new_payee_bin, ratio, prior_avg, prior_dec, prob, decision_text, reasoning_steps, judgment_text, counterfactual_text, cf_results, meta_html, source_label): """5세대 카드 HTML 렌더링""" dec, bg, fg = decide(prob) context_html = ( f"[SYS] 당신은 한국 은행의 FDS 분석 전문가입니다. 거래 정보·1-4세대 사전 판단·XAI 결과를 토대로 종합 판정하세요.
" f"[INPUT] amount={amount:.0f}만, hour={int(hour):02d}, " f"new_payee={'true' if new_payee_bin==1 else 'false'}, ratio={ratio:.1f}×
" f"[PRIOR] 1-4세대 평균: {prior_avg*100:.1f}% / 종합: {prior_dec}
" f"[XAI-IN] 3세대 SHAP + 4세대 GNNExplainer 결과 함께 주입 (Grounded Reasoning)
" f"[TASK] JSON 형식으로 risk_score, decision, reasoning_steps(+evidence_source), judgment, counterfactual 출력" ) cot_rows = "" for i, step in enumerate(reasoning_steps, 1): if isinstance(step, dict): step_text = step.get("step", "") attn = step.get("attention", 0.0) evidence = step.get("evidence_source", "-") else: step_text = str(step) attn = 0.0 evidence = "-" try: attn = float(attn) except (ValueError, TypeError): attn = 0.0 # evidence_source에 따라 배지 색상 다르게 if "SHAP" in evidence or "3세대" in evidence: ev_bg, ev_fg = "#FAEEDA", "#854F0B" elif "GNN" in evidence or "4세대" in evidence: ev_bg, ev_fg = "#E6F1FB", "#0C447C" elif "도메인" in evidence or "지식" in evidence: ev_bg, ev_fg = "#FAECE7", "#993C1D" else: ev_bg, ev_fg = "#F1EFE8", "#5F5E5A" cot_rows += ( f"{i}" f"{step_text}" f"{evidence}" f"{attn:.2f}" ) judg_color = "#3B6D11" if prob < 0.5 else "#633806" judgment_html = ( f"{decision_text} 권고 (의심도 {prob*100:.0f}%)

" f"{judgment_text}" ) # ─── Counterfactual 시각화 ───────────────────────────── cf_rows = "" for item in cf_results: # 새 시그니처: (label, cf_prob, prob_drop, flipped, logit_drop) if len(item) == 5: label, cf_prob, prob_drop, flipped, logit_drop = item else: # 이전 시그니처 호환 label, cf_prob, prob_drop, flipped = item logit_drop = 0.0 flip_badge = ("판정 뒤집힘 ✓" if flipped else "") # logit drop이 의미있는 신호 (saturate되어도 살아남음) logit_color = "#3B6D11" if logit_drop > 0.5 else "#888" prob_color = "#3B6D11" if prob_drop > 0.05 else "#888" cf_rows += ( f"" f"{label}{flip_badge}" f"{cf_prob:.4f}" f"−{prob_drop:.4f}" f"−{logit_drop:.4f}" f"" ) cf_table = f""" {cf_rows}
반사실 가정 변경 후 P ΔP Δ logit

※ saturate 영역(P≈1.0)에서는 ΔP가 작아도 Δ logit이 크면 실제로는 강하게 정상 쪽으로 끌어당기는 변경임

""" cf_natural = "" if counterfactual_text: cf_natural = ( f"
" f"💬 LLM이 생성한 반사실 설명:
{counterfactual_text}
" ) gen5_setup = feature_setup_box( actor_label="프롬프트만 사람, 추론은 전적으로 모델", actor_color='model', items=[ ("시스템 프롬프트", "사람", "'당신은 FDS 분석가입니다' 등의 역할 부여만 사람이 작성"), ("학습 데이터", "모델", "Anthropic이 인터넷 규모 데이터로 사전 학습 (수조 토큰)"), ("도메인 지식", "모델", "보이스피싱 패턴, 부동산 거래 정형 등을 사전 학습으로 보유"), ("추론 단계 (CoT)", "모델", "각 단계에서 무엇에 주목할지 모델이 스스로 결정"), ("최종 판단 문장", "모델", "자연어로 자동 생성"), ("판정 임계값", "사람", "0.5(추가인증) / 0.7(차단)"), ], explanation="모델이 사전 학습된 도메인 지식으로 '왜 사기인지/아닌지'를 자연어로 추론. 4세대까지의 250건 학습과는 차원이 다른 규모의 사전 학습이 깔려 있음." ) # ─── XAI 구현 박스 (5세대) — 핵심 변화 ────────────────── gen5_xai = xai_box( method="CoT + Tool Use + Grounded Reasoning + Counterfactual", status='generative', era="2023년 ChatGPT 이후 ~ 현재", items=[ ("패러다임 전환", "3·4세대: 모델이 판단 후 → 별도 알고리즘이 설명 추출 (Post-hoc). 5세대: 추론 과정 자체가 자연어로 생성되어 그게 곧 설명 (Generative)"), ("Chain-of-Thought (CoT)", "각 추론 단계를 모델이 직접 자연어로 출력. 위 '추론 체인' 표가 그것. 어느 신호에 얼마나 주목했는지(attention)까지 함께 산출"), ("Tool Use로 구조 강제", "submit_fds_judgment 도구의 JSON 스키마로 출력 형식을 강제 → 자유 텍스트 파싱 오류 원천 차단, DB 적재·감사 추적 가능"), ("Grounded Reasoning", "3세대 SHAP과 4세대 GNNExplainer 결과를 프롬프트에 함께 주입 → LLM이 단순 자체 추론이 아니라 하위 모델의 설명까지 참조. evidence_source 필드로 출처 추적"), ("Counterfactual Explanation", "'이 거래가 통과되려면 무엇이 달라져야 했나'를 자연어로 생성 + GNN으로 빠르게 검증 (아래 표). 고객 응대·이의 제기 처리에 직접 활용 가능"), ("Faithfulness 한계", "LLM이 출력한 reasoning이 실제 내부 계산을 정확히 반영한다는 보장은 학계 미해결 문제 → 그래서 1-4세대 점수와 교차 검증하는 하이브리드가 권장됨"), ], takeaway="5세대 XAI의 본질: '설명이 사후 추출되는 무엇'에서 '판단의 산출물 그 자체'로 변화. SHAP·GNNExplainer는 사라지지 않고 LLM의 입력 컨텍스트로 흡수되어 함께 작동." ) return f"""
{card_header("GEN 5 · FOUNDATION MODEL (LLM)", f"초거대 추론 모델 ({source_label})", dec, bg, fg, f"의심도 {prob*100:.0f}%")} {gen5_setup} {gen5_xai}

컨텍스트 토큰화 (1-4세대 XAI 결과 함께 주입)

{context_html}

🧠 추론 체인 (Chain-of-Thought) + 근거 출처 (evidence_source)

{cot_rows}
# 추론 내용 근거 출처 Attention

📝 생성된 자연어 판단

{judgment_html}

🔄 Counterfactual Explanation (반사실 설명)

"무엇이 달라졌다면 이 거래가 통과되었을까?" — 4세대 GNN으로 빠르게 검증한 결과

{cf_natural} {cf_table}
생성 파라미터 보기
{meta_html}

강점: 1-4세대 XAI 결과를 컨텍스트로 받아 자연어 추론 + 권고 조치 + 반사실 설명까지 한 번에 생성

""" def render_gen5_simulation(amount, hour, new_payee_bin, ratio, prior_avg, prior_dec, gen3_shap, gen4_edges, gen4_feats, cf_results): """API 호출 실패/미설정 시 fallback용 시뮬레이션""" is_high_risk = (new_payee_bin == 1 and (hour <= 6 or hour >= 22) and amount >= 500) prob = 0.95 if is_high_risk else min(prior_avg + 0.02, 0.98) payee_str = "신규 수취인" if new_payee_bin == 1 else "기존 수취인" time_str = "새벽" if (hour <= 6 or hour >= 22) else "일반" # SHAP/GNN XAI 결과에서 가장 큰 신호 추출하여 reasoning에 반영 top_shap = max(gen3_shap, key=lambda x: abs(x[1])) if gen3_shap else ("금액", 0) top_edge = max(gen4_edges, key=lambda x: abs(x[2])) if gen4_edges else ("이웃", 0, 0) reasoning_steps = [ { "step": f"4세대 GNNExplainer: '{top_edge[0]}' 엣지 기여도 {top_edge[2]:+.3f} — 그래프 구조상 핵심 신호", "attention": 0.32, "evidence_source": "4세대 GNNExplainer" }, { "step": f"3세대 SHAP: '{top_shap[0]}' 기여도 {top_shap[1]:+.3f} — 평균 거래 대비 위험 방향", "attention": 0.28, "evidence_source": "3세대 SHAP" }, { "step": f"시간 {int(hour):02d}시 + {payee_str} → 보이스피싱 정형 패턴 매칭", "attention": 0.24, "evidence_source": "도메인 지식" }, { "step": f"평소 {ratio:.1f}배 금액 → 송금인 평소 행동 이탈도 측정", "attention": 0.16, "evidence_source": "거래 컨텍스트" }, ] if is_high_risk: decision_text = "차단" judgment_text = ( f"{int(hour):02d}시 {time_str} 시간대에 평소보다 {ratio:.1f}배 급증한 {amount:.0f}만원이 " f"{payee_str}에게 이체되는 것은 전형적인 보이스피싱 패턴입니다. " f"3세대 SHAP에서도 '{top_shap[0]}'이 가장 강한 위험 신호로 나타났고, " f"4세대 GNNExplainer 결과 '{top_edge[0]}'이 그래프 구조 차원에서 핵심 기여를 했습니다.

" f"권고 조치: ① 즉시 거래 보류, ② 등록된 전화번호로 본인 직접 확인, ③ 확인 전 자금 동결 24시간 유지" ) counterfactual_text = "수취인이 송금인의 기존 거래 이력이 있는 계좌였거나, 거래 시간이 주간(9-18시)이었다면 위험도가 큰 폭으로 감소했을 것입니다." elif prob >= 0.5: decision_text = "추가 인증" judgment_text = ( f"{int(hour):02d}시 거래에서 일부 이상 신호({ratio:.1f}배 금액, {payee_str})가 감지되었으나 " f"결정적 위험 패턴은 아닙니다. 차단보다는 추가 인증으로 본인 의사를 확인하는 것이 적절합니다.

" f"권고 조치: ① ARS 또는 OTP 추가 인증, ② 송금 의도 재확인 메시지 발송" ) counterfactual_text = "신규 수취인 플래그가 없었거나 금액이 평소 수준이었다면 통과 가능했을 것입니다." else: decision_text = "통과" judgment_text = ( f"{int(hour):02d}시 거래의 패턴이 송금인의 평소 행동 범위 내에 있으며, " f"1-4세대 모델 모두 위험 신호를 강하게 보내지 않았습니다. 정상 거래로 판단됩니다.

" f"권고 조치: 별도 조치 없이 거래 진행" ) counterfactual_text = "(통과 거래이므로 반사실 분석은 적용 대상 아님)" meta_html = ( "mode: SIMULATION (API 미연결)
" "이 결과는 if-else 하드코딩으로 생성된 시뮬레이션입니다.
" "단, reasoning_steps에는 실제 3세대 SHAP과 4세대 GNNExplainer 결과가 반영됨.
" "실제 Claude 호출을 활성화하려면 ANTHROPIC_API_KEY 환경변수를 설정하세요." ) return render_gen5_card( amount, hour, new_payee_bin, ratio, prior_avg, prior_dec, prob, decision_text, reasoning_steps, judgment_text, counterfactual_text, cf_results, meta_html, source_label="시뮬레이션 모드" ) def render_gen5(amount, hour, new_payee_bin, ratio, prior_avg, use_claude_api=True): """5세대 진입점. 1-4세대 XAI 결과(SHAP, GNNExplainer)를 모두 수집하여 LLM에 함께 전달. """ prior_dec, _, _ = decide(prior_avg) # 3·4세대 XAI 결과 수집 (Grounded Reasoning용) gen3_shap_vals, _ = compute_shap_values_gen3(amount, hour, new_payee_bin, ratio) gen3_shap = list(zip(FEATURES, gen3_shap_vals)) _, gen4_edges = compute_gnn_edge_attribution(amount, hour, new_payee_bin, ratio) _, gen4_feats = compute_gnn_node_feature_attribution(amount, hour, new_payee_bin, ratio) # Counterfactual 후보 평가 (GNN으로 빠르게) cf_results = build_counterfactual_gen5(amount, hour, new_payee_bin, ratio) if not use_claude_api or not CLAUDE_AVAILABLE: return render_gen5_simulation( amount, hour, new_payee_bin, ratio, prior_avg, prior_dec, gen3_shap, gen4_edges, gen4_feats, cf_results ) parsed, meta_or_err = call_claude_api( amount, hour, new_payee_bin, ratio, prior_avg, prior_dec, gen3_shap=gen3_shap, gen4_edges=gen4_edges, gen4_feats=gen4_feats ) if parsed is None: fallback_html = render_gen5_simulation( amount, hour, new_payee_bin, ratio, prior_avg, prior_dec, gen3_shap, gen4_edges, gen4_feats, cf_results ) warning = ( f'
' f'⚠️ Claude API 호출 실패. 시뮬레이션 모드로 대체합니다.
' f'사유: {meta_or_err}' f'
' ) return warning + fallback_html try: prob = float(parsed.get("risk_score", 0.5)) prob = max(0.0, min(1.0, prob)) decision_text = parsed.get("decision", "추가 인증") reasoning_steps = parsed.get("reasoning_steps", []) judgment_text = parsed.get("judgment", "(판단 내용 누락)") counterfactual_text = parsed.get("counterfactual", "") meta = meta_or_err meta_html = ( f"model: {meta['model']} / temperature: 0.2 / max_tokens: 1024
" f"input_tokens: {meta['input_tokens']} / output_tokens: {meta['output_tokens']} / " f"latency: {meta['latency']:.2f}s
" f"mode: LIVE API CALL ✓ · Grounded with 3-4세대 XAI" ) return render_gen5_card( amount, hour, new_payee_bin, ratio, prior_avg, prior_dec, prob, decision_text, reasoning_steps, judgment_text, counterfactual_text, cf_results, meta_html, source_label=f"실제 {CLAUDE_MODEL}" ) except (KeyError, ValueError, TypeError) as e: fallback_html = render_gen5_simulation( amount, hour, new_payee_bin, ratio, prior_avg, prior_dec, gen3_shap, gen4_edges, gen4_feats, cf_results ) warning = ( f'
' f'⚠️ Claude 응답 파싱 실패. 시뮬레이션 모드로 대체합니다.
' f'' f'{type(e).__name__}: {str(e)[:150]}
' ) return warning + fallback_html # ============================================================ # 4. 메인 분석 함수 # ============================================================ def render_xai_evolution_summary(): """모든 카드 위에 표시되는 XAI 진화 요약 박스 (강의용)""" return """

🔍 세대별 XAI 구현 방식 — 한눈에 보기

세대 XAI 기법 설명 단위 한 줄 핵심
1세대 불필요 (Self-Explanatory) 발동된 룰 룰북 자체가 설명. 별도 알고리즘 필요 없음
2세대 계수 분해 (Coefficient) 피처별 wᵢ·xᵢ 선형 모델 = 정확한 SHAP. 내재적 설명력 유지
3세대 ⚡ TreeSHAP (2017~) 피처 기여도 XAI가 본격 등장한 세대. 모델·설명기 분리. 현재 금융권 표준
4세대 GNNExplainer (2019~) 엣지 + 노드 피처 설명 단위가 '그래프 구조'로 확장. 관계망 기반 설명
5세대 🆕 CoT + Tool Use + Counterfactual 자연어 추론 체인 패러다임 전환: 설명이 사후 추출 → 판단의 산출물 그 자체. 3·4세대 XAI를 컨텍스트로 흡수

💡 강의 포인트: XAI가 '문제'가 된 것은 3세대(XGBoost)부터. 1-2세대는 내재적 설명력이, 5세대는 생성형 설명이 있어 별도 알고리즘이 덜 중요. 3-4세대에서 SHAP/GNNExplainer가 핵심.

""" def analyze_transaction(amount, hour, new_payee, ratio, use_claude_api): start_time = time.time() new_payee_bin = 1 if new_payee == "예" else 0 # 1세대 g1 = render_gen1(amount, hour, new_payee_bin, ratio) _, score1 = evaluate_gen1(amount, hour, new_payee_bin, ratio) prob1 = min(score1 / 100, 0.99) # 2세대 g2 = render_gen2(amount, hour, new_payee_bin, ratio) input_vec = np.array([amount, hour, new_payee_bin, ratio], dtype=float) logit2 = (GEN2_COEF * input_vec).sum() + GEN2_INTERCEPT prob2 = float(1 / (1 + np.exp(-logit2))) # 3세대 g3 = render_gen3(amount, hour, new_payee_bin, ratio) input_df = pd.DataFrame([[amount, hour, new_payee_bin, ratio]], columns=FEATURES) prob3 = float(gen3_model.predict_proba(input_df)[0][1]) # 4세대 prob4, g4 = render_gen4(amount, hour, new_payee_bin, ratio, prob3) # 5세대 (1-4세대 평균을 prior로) - 토글에 따라 API/시뮬레이션 분기 prior_avg = (prob1 + prob2 + prob3 + prob4) / 4 g5 = render_gen5(amount, hour, new_payee_bin, ratio, prior_avg, use_claude_api=use_claude_api) elapsed = time.time() - start_time # 모드 배지 if use_claude_api and CLAUDE_AVAILABLE: mode_badge = ('' '⚡ 5세대 LIVE API 모드') elif use_claude_api and not CLAUDE_AVAILABLE: mode_badge = ('' '⚠️ API 키 미설정 → 시뮬레이션 자동 전환') else: mode_badge = ('' '🧪 5세대 시뮬레이션 모드 (토글 OFF)') summary = f"""

분석 대상 거래

{mode_badge}

금액

{amount:.0f}만원

거래 시간

{int(hour):02d}시

신규 수취인

{new_payee}

과거 대비 배율

{ratio:.1f}×

분석 소요시간: {elapsed:.3f}초

""" return summary + render_xai_evolution_summary() + g1 + g2 + g3 + g4 + g5 # ============================================================ # 5. Gradio UI # ============================================================ with gr.Blocks(theme=gr.themes.Default(), title="FDS 1-5세대 비교 데모") as demo: _api_badge = ( f'' f'● Claude API 연결됨 ({CLAUDE_MODEL})' if CLAUDE_AVAILABLE else '' '○ API 미연결 (5세대 시뮬레이션 모드)' ) gr.HTML(f"""

🛡️ FDS 1-5세대 비교 데모

실무 운영 담당자 시연용 · 판단 수식·가중치·근거 + 세대별 XAI 구현 전체 노출 모드

{_api_badge}
""") with gr.Row(): with gr.Column(scale=1): amount_in = gr.Number(label="금액 (만원)", value=700) hour_in = gr.Slider(label="거래 시간 (0-23시)", minimum=0, maximum=23, value=3, step=1) payee_in = gr.Radio(label="신규 수취인", choices=["아니오", "예"], value="예") ratio_in = gr.Number(label="과거 대비 배율", value=14.0) use_api_in = gr.Checkbox( label="🤖 5세대에 실제 Claude API 호출", value=CLAUDE_AVAILABLE, interactive=CLAUDE_AVAILABLE, info=( "체크: claude-sonnet-4-6 실제 호출 (지연 1~3초, 호출당 약 10원)" if CLAUDE_AVAILABLE else "ANTHROPIC_API_KEY가 설정되지 않아 시뮬레이션만 가능합니다" ) ) submit_btn = gr.Button("🔍 분석 실행", variant="primary") gr.Examples( examples=[ [700, 3, "예", 14.0], [800, 23, "예", 8.0], [45, 2, "아니오", 1.2], [1500, 14, "아니오", 2.0], [499, 23, "예", 4.5], ], inputs=[amount_in, hour_in, payee_in, ratio_in], label="시연 예시 (마지막은 1세대 룰을 회피하는 케이스)" ) with gr.Column(scale=2): output_html = gr.HTML( "
" "좌측에서 거래 조건을 설정하고 [분석 실행] 버튼을 누르세요." "
" ) submit_btn.click( fn=analyze_transaction, inputs=[amount_in, hour_in, payee_in, ratio_in, use_api_in], outputs=output_html ) if __name__ == "__main__": demo.launch(ssr_mode=False)