import sys
# 1. 최상단 몽키 패치 - 모든 임포트보다 최우선
import huggingface_hub
if not hasattr(huggingface_hub, "HfFolder"):
class MockHfFolder:
@staticmethod
def get_token(): return None
@staticmethod
def save_token(token): pass
@staticmethod
def delete_token(): pass
huggingface_hub.HfFolder = MockHfFolder
import gradio as gr
import os
import re
import zipfile
import tempfile
import numpy as np
import pandas as pd
import tensorflow as tf
import joblib
import traceback
from huggingface_hub import hf_hub_download
# TF 최적화 경고 방지 및 안정성
os.environ['TF_ENABLE_ONEDNN_OPTS'] = '0'
# 무거운 모델 지연 로딩
_models = {"predictor": None, "consultant": None}
REPO_ID = "dev-yuje/gardio_test"
def load_keras_model_compat(model_path):
"""quantization_config 역직렬화 오류를 config.json 패치로 우회"""
with tempfile.TemporaryDirectory() as tmpdir:
with zipfile.ZipFile(model_path, 'r') as z:
z.extractall(tmpdir)
config_path = os.path.join(tmpdir, 'config.json')
with open(config_path, 'r', encoding='utf-8') as f:
config_str = f.read()
config_str = re.sub(r',\s*"quantization_config":\s*null', '', config_str)
config_str = re.sub(r'"quantization_config":\s*null,?\s*', '', config_str)
with open(config_path, 'w', encoding='utf-8') as f:
f.write(config_str)
fixed_path = model_path + '.tmp_fixed.keras'
with zipfile.ZipFile(fixed_path, 'w', zipfile.ZIP_DEFLATED) as z:
for root, dirs, files in os.walk(tmpdir):
for file in files:
fp = os.path.join(root, file)
arcname = os.path.relpath(fp, tmpdir)
z.write(fp, arcname)
model = tf.keras.models.load_model(fixed_path, compile=False)
os.remove(fixed_path)
return model
def load_all_models():
if _models["predictor"] is None:
try:
# 내부 예측 로직 (Self-contained)
class RobustCreditPredictor:
def __init__(self):
self.preprocessor_path = "models/preprocessor.pkl"
self.model_path = "models/telecom_cb_model.keras"
self.preprocessor = None
self.model = None
self.load_error = "초기화됨"
self.load_resources()
def load_resources(self):
try:
# [개선] 로컬 파일 우선 로드 → 없으면 HuggingFace 다운로드
# --- 전처리기 로드 ---
try:
if os.path.exists(self.preprocessor_path):
self.preprocessor = joblib.load(self.preprocessor_path)
print(f"✅ 전처리기 로컬 로드 성공: {self.preprocessor_path}")
else:
cached_preprocessor = hf_hub_download(repo_id=REPO_ID, filename=self.preprocessor_path, repo_type="space")
self.preprocessor = joblib.load(cached_preprocessor)
print(f"✅ 전처리기 HuggingFace 다운로드 성공")
except Exception as prep_e:
self.load_error = f"전처리기 로드 실패: {prep_e}"
return
# --- 모델 로드 ---
try:
if os.path.exists(self.model_path):
target_path = self.model_path
print(f"✅ 모델 로컬 경로 사용: {target_path}")
else:
target_path = hf_hub_download(repo_id=REPO_ID, filename=self.model_path, repo_type="space")
print(f"✅ 모델 HuggingFace 다운로드 완료")
fsize = os.path.getsize(target_path)
if fsize < 1000:
self.load_error = f"파일이 너무 작음({fsize}B). LFS 포인터일 가능성 있음."
return
self.model = load_keras_model_compat(target_path)
self.load_error = "성공"
print(f"✅ 모델 로드 성공 (파일 크기: {fsize:,}B)")
except Exception as model_e:
self.load_error = f"모델 로드 실패: {str(model_e)[:300]}"
except Exception as e:
self.load_error = f"리소스 로드 통합 에러: {e}"
def predict(self, features_dict):
try:
if self.model is None or self.preprocessor is None:
err_short = self.load_error[:500] if self.load_error else "원인 불명"
return f"Error: 로드 상태 확인 요망. 원본 에러: {err_short}"
ALL_FEATURES = [
'C1Z001386', 'C1M210000', 'C18210000', 'C1L120001', 'C1L120004',
'L10210000', 'L90210100', 'L90210200', 'L10210B00', 'L10216000',
'L10217000', 'D10110000', 'D10133000', 'PERF1'
]
input_values = [float(features_dict.get(col, 0.0)) for col in ALL_FEATURES]
df = pd.DataFrame([input_values], columns=ALL_FEATURES)
log_cols = ['C1Z001386', 'C1L120004', 'D10110000', 'D10133000', 'L90210200',
'L10216000', 'L10210B00', 'L10217000', 'L90210100', 'L10210000']
df[log_cols] = np.log1p(df[log_cols].astype(float).clip(lower=0))
scaled_data = self.preprocessor.transform(df)
prediction = self.model.predict(scaled_data, verbose=0)
return float(prediction[0][0])
except Exception as e:
return f"Error: 예측 연산 에러: {str(e)}"
_models["predictor"] = RobustCreditPredictor()
# 상담사 로직
from langchain_google_genai import ChatGoogleGenerativeAI
class Consultant:
def __init__(self):
api_key = os.getenv("GOOGLE_API_KEY", "")
from config import LLM_MODEL
# [수정] 모델 명칭 및 API 버전 호환성 고려 (config 연동)
self.llm = ChatGoogleGenerativeAI(
model=LLM_MODEL,
google_api_key=api_key,
temperature=0.7,
convert_system_message_to_human=True
)
self.embedding_model = None
self.retriever = None
def lazy_load_search(self):
if self.embedding_model is None:
try:
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from config import EMBEDDING_MODEL, FAISS_PATH, RETRIEVER_K
self.embedding_model = HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL)
if os.path.exists(FAISS_PATH):
self.vectorstore = FAISS.load_local(FAISS_PATH, self.embedding_model, allow_dangerous_deserialization=True)
self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": RETRIEVER_K})
except: pass
_models["consultant"] = Consultant()
except Exception as e:
print(f"Grand Load Error: {e}")
FEATURES_DETAIL = {
'C1Z001386': ('1년 내 카드 총 이용금액', '만원 단위', '0'),
'C1M210000': ('보유 신용카드 수', '개수', '0'),
'C18210000': ('보유 체크카드 수', '개수', '0'),
'C1L120001': ('카드 총 한도금액', '만원 단위', '0'),
'C1L120004': ('신용카드 개설 후 경과일수', '일 단위', '0'),
'L90210100': ('은행업종 대출 건수', '개수', '0'),
'L90210200': ('카드업종 대출 건수', '개수', '0'),
'L10210B00': ('보험업종 대출 건수', '개수', '0'),
'L10216000': ('신용대출 건수', '개수', '0'),
'L10217000': ('담보대출 건수', '개수', '0'),
'D10110000': ('과거 연체 건수', '개수', '0'),
'D10133000': ('총 연체 상환 금액', '만원 단위', '0'),
'PERF1': ('1년 내 90일 이상 연체 경험', '체크 시 연체 경험 있음', None),
}
ALL_KEYS = list(FEATURES_DETAIL.keys())
def get_grade_info(score_str):
"""점수 → (등급, 등급 구분, 의미/특징, 색상) 반환"""
try:
score = int(score_str)
except:
return None
if score >= 942: grade, label, desc, color = "1등급", "최우량등급", "오랜 신용거래 경력과 다양하고 우량한 신용거래 실적을 보유하고 있어 부실화 가능성이 매우 낙음", "#A8D8EA"
elif score >= 891: grade, label, desc, color = "2등급", "최우량등급", "오랜 신용거래 경력과 다양하고 우량한 신용거래 실적을 보유하고 있어 부실화 가능성이 매우 낙음", "#A8D8EA"
elif score >= 832: grade, label, desc, color = "3등급", "우량등급", "활발한 신용거래 실적은 없으나, 꼸준히 우량 거래를 지속한다면 상위등급 진입 가능하며 부실화 가능성은 낙은 수준임", "#B8F0C8"
elif score >= 768: grade, label, desc, color = "4등급", "우량등급", "활발한 신용거래 실적은 없으나, 꼸준히 우량 거래를 지속한다면 상위등급 진입 가능하며 부실화 가능성은 낙은 수준임", "#B8F0C8"
elif score >= 698: grade, label, desc, color = "5등급", "일반등급", "비교적 금리가 높은 금융업권과의 거래가 있는 고객으로, 단기연체 경험이 있으며 부실화 가능성은 일반적인 수준임", "#FEE8A0"
elif score >= 620: grade, label, desc, color = "6등급", "일반등급", "비교적 금리가 높은 금융업권과의 거래가 있는 고객으로, 단기연체 경험이 있으며 부실화 가능성은 일반적인 수준임", "#FEE8A0"
elif score >= 530: grade, label, desc, color = "7등급", "주의등급", "비교적 금리가 높은 금융업권과의 거래가 많은 고객으로, 단기연체 경험을 비교적 많이 보유하고 있어 부실화 가능성이 높음", "#FFCB9A"
elif score >= 454: grade, label, desc, color = "8등급", "주의등급", "비교적 금리가 높은 금융업권과의 거래가 많은 고객으로, 단기연체 경험을 비교적 많이 보유하고 있어 부실화 가능성이 높음", "#FFCB9A"
elif score >= 335: grade, label, desc, color = "9등급", "위험등급", "현재 연체 중이거나 매우 심각한 연체 경험을 보유하고 있어 부실화 가능성이 매우 높음", "#FFB3B3"
else: grade, label, desc, color = "10등급", "위험등급", "현재 연체 중이거나 매우 심각한 연체 경험을 보유하고 있어 부실화 가능성이 매우 높음", "#FFB3B3"
return grade, label, desc, color
SCORE_PLACEHOLDER = '''
📋
📌 정보를 입력하고 점수 분석하기 버튼을 눌러주세요
'''
def make_score_html(score_str):
if not score_str or score_str.startswith("❌") or score_str.startswith("⚠️"):
return f'''
{score_str}
'''
info = get_grade_info(score_str)
if info is None:
return f'
{score_str}
'
grade, label, desc, color = info
# 다크모드 호환: 배경 반투명, 텍스트는 CSS inherit 대신 명시적 다크모드 공유 새문 (prefers-color-scheme)
return f'''
{score_str}점
{grade} | {label}
{desc}
'''
def handle_predict(*args):
try:
load_all_models()
features_dict = {}
for i, key in enumerate(ALL_KEYS):
val_raw = str(args[i]).strip().replace(",", "")
if key == 'PERF1':
features_dict[key] = 1.0 if (val_raw.lower() == 'true' or val_raw == '1' or args[i] is True) else 0.0
else:
try:
features_dict[key] = float(val_raw or 0)
except:
return f"❌ 오류: '{FEATURES_DETAIL[key][0]}' 숫자 아님", "❌"
features_dict['L10210000'] = features_dict['L10216000'] + features_dict['L10217000']
res = _models["predictor"].predict(features_dict)
if isinstance(res, str) and "Error" in res:
return f"❌ 분석 실패: {res}", make_score_html(f"❌ 로드 실패: {res[:60]}")
score_val = str(int(round(float(res))))
return {"features": features_dict, "score": score_val}, make_score_html(score_val)
except Exception as e:
return f"❌ 시스템 오류: {str(e)}", make_score_html(f"❌ 시스템 에러")
def generate_response(chatbot, user_message, analysis_report):
if not user_message: yield chatbot, ""; return
# [최종 수정] Gradio 5의 'Data incompatible' 에러 해결을 위해 명시적인 딕셔너리 포맷 사용
chatbot.append({"role": "user", "content": user_message})
chatbot.append({"role": "assistant", "content": "🔍 [Retrieval] 관련 규정 및 지침을 검색하고 있습니다..."})
yield chatbot, ""
try:
load_all_models()
cons = _models["consultant"]
cons.lazy_load_search()
context = ""
if cons.retriever:
try:
docs = cons.retriever.invoke(user_message)
context = "\n\n".join([d.page_content for d in docs])
except: pass
chatbot[-1] = {"role": "assistant", "content": "🧠 [Augmentation] 검색된 지식을 바탕으로 프롬프트를 구성하고 있습니다..."}
yield chatbot, ""
from llm.prompt import QA_PROMPT
if isinstance(analysis_report, dict) and "features" in analysis_report:
score_val = analysis_report.get("score", "미측정")
# L10210000은 파생 컨럼으로 FEATURES_DETAIL에 없음 - 키가 있는 것만 표시
features_text = "\n".join([f"- {FEATURES_DETAIL[k][0]}: {v}" for k, v in analysis_report["features"].items() if k in FEATURES_DETAIL])
query_text = f"■ 고객 신용 점수: {score_val}점\n■ 고객의 현재 상태(입력된 정보):\n{features_text}\n\n■ 고객 질문: {user_message}"
else:
query_text = f"■ 고객 신용 점수: 정보 없음(일반 질문 상태)\n■ 상태: 분석을 진행하지 않음\n■ 질문: {user_message}"
full_prompt = QA_PROMPT.format(context=context, query=query_text)
# [중요] 딕셔너리 리스트로 전달 (Gradio 5 UI 대응)
from langchain_core.messages import HumanMessage
messages = [HumanMessage(content=full_prompt)]
chatbot[-1] = {"role": "assistant", "content": "💡 [Generation] 답변을 생성 중입니다...\n\n"}
yield chatbot, ""
answer_buffer = ""
for chunk in cons.llm.stream(messages):
answer_buffer += chunk.content
# 마지막 메시지의 내용을 딕셔너리 포맷으로 업데이트
chatbot[-1] = {"role": "assistant", "content": answer_buffer}
yield chatbot, ""
except Exception as e:
chatbot[-1] = {"role": "assistant", "content": f"⚠️ 상담 에러: {str(e)}"}
yield chatbot, ""
with gr.Blocks(title="KCB AI Consultant") as demo:
analysis_report = gr.State(None)
gr.Markdown("# 🛡️ KCB AI 신용 상담 시스템")
with gr.Row(equal_height=False):
# ==== 왼쪽: 입력 패널 (2열 그리드) ====
with gr.Column(scale=1, min_width=340):
gr.HTML('
📊 신용 지표 입력
')
input_list = []
keys_no_perf = [k for k in ALL_KEYS if k != 'PERF1']
# 2열로 입력 필드 배치
for i in range(0, len(keys_no_perf), 2):
with gr.Row():
for key in keys_no_perf[i:i+2]:
field_label, unit, _ = FEATURES_DETAIL[key]
unit_short = unit.replace(" 단위", "").replace("개수", "개")
# label에 단위를 괴호로 붙이되, info 없애 레이아웃 정리
tb = gr.Textbox(
label=f"{field_label} ({unit_short})",
placeholder="0",
min_width=120
)
input_list.append(tb)
# PERF1 켬 코너로 배치
with gr.Row():
perf_cb = gr.Checkbox(label=FEATURES_DETAIL['PERF1'][0], info=FEATURES_DETAIL['PERF1'][1], value=False)
input_list.append(perf_cb)
predict_btn = gr.Button("📈 점수 분석하기", variant="primary", size="lg")
# ==== 오른쪽: 결과 패널 ====
with gr.Column(scale=2):
gr.HTML('