Upload app.py

app.py

@@ -0,0 +1,284 @@
+# -*- coding: utf-8 -*-
+"""app.ipynb
+
+Automatically generated by Colab.
+
+Original file is located at
+    https://colab.research.google.com/drive/1y3yISz14Lpsr131OIJCKA77lwbFmEJzB
+"""
+
+import streamlit as st
+import os
+import joblib
+import torch
+import numpy as np
+import html
+from transformers import AutoTokenizer, AutoModel, logging as hf_logging
+
+# Hugging Face Transformers 로깅 레벨 설정 (오류만 표시)
+hf_logging.set_verbosity_error()
+
+# ────────── 설정 (Hugging Face Spaces 환경에 맞게 조정) ──────────
+MODEL_NAME = "bert-base-uncased"
+DEVICE     = "cpu"  # Hugging Face Spaces 무료 티어는 CPU 사용
+SAVE_DIR   = "저장저장1" # 업로드할 폴더명과 일치해야 함
+LAYER_ID   = 4      # 원본 코드의 SeparationScore 최고 레이어
+SEED       = 0      # 원본 코드의 SEED 값
+CLF_NAME   = "linear" # 원본 코드의 CLF_NAME
+
+# ────────── 모델 로드 (Streamlit 캐시 사용으로 앱 전체에서 한 번만 실행) ──────────
+@st.cache_resource
+def load_all_models_and_data():
+    """
+    LDA, 분류기, 토크나이저, BERT 모델 및 관련 행렬들을 로드합니다.
+    Hugging Face Spaces에 배포 시 파일 경로가 정확해야 합니다.
+    """
+    lda_file_path = os.path.join(SAVE_DIR, f"lda_layer{LAYER_ID}_seed{SEED}.pkl")
+    clf_file_path = os.path.join(SAVE_DIR, f"{CLF_NAME}_layer{LAYER_ID}_projlda_seed{SEED}.pkl")
+
+    # 파일 존재 여부 확인 (배포 환경 디버깅용)
+    if not os.path.isdir(SAVE_DIR):
+        st.error(f"오류: 모델 저장 디렉토리 '{SAVE_DIR}'를 찾을 수 없습니다. Spaces에 폴더가 올바르게 업로드되었는지, 이름이 일치하는지 확인하세요.")
+        return None
+    if not os.path.exists(lda_file_path):
+        st.error(f"오류: LDA 모델 파일 '{lda_file_path}'를 찾을 수 없습니다. 파일 이름과 경로를 확인하세요.")
+        return None
+    if not os.path.exists(clf_file_path):
+        st.error(f"오류: 분류기 모델 파일 '{clf_file_path}'를 찾을 수 없습니다. 파일 이름과 경로를 확인하세요.")
+        return None
+
+    try:
+        lda = joblib.load(lda_file_path)
+        clf = joblib.load(clf_file_path)
+    except Exception as e:
+        st.error(f"모델 파일 로드 중 오류 발생: {e}")
+        st.error("파일이 손상되었거나, joblib 버전 호환성 문제가 있을 수 있습니다.")
+        return None
+
+    if hasattr(clf, "base_estimator"):  # Calibrated Ridge 경우
+        clf = clf.base_estimator
+
+    # LDA 행렬·평균, 분류기 가중치를 PyTorch Tensor로 변환
+    W_tensor   = torch.tensor(lda.scalings_,  dtype=torch.float32, device=DEVICE)
+    mu_vector  = torch.tensor(lda.xbar_,     dtype=torch.float32, device=DEVICE)
+    w_p_tensor = torch.tensor(clf.coef_,     dtype=torch.float32, device=DEVICE)
+    b_p_vector = torch.tensor(clf.intercept_, dtype=torch.float32, device=DEVICE)
+
+    # Hugging Face 토크나이저 및 BERT 모델 로드
+    try:
+        tokenizer_obj = AutoTokenizer.from_pretrained(MODEL_NAME, use_fast=True)
+        model_obj     = AutoModel.from_pretrained(
+            MODEL_NAME, output_hidden_states=True
+        ).to(DEVICE).eval()
+    except Exception as e:
+        st.error(f"Hugging Face 모델 ({MODEL_NAME}) 로드 중 오류: {e}")
+        st.error("인터넷 연결 또는 모델 이름이 올바른지 확인하세요.")
+        return None
+
+    # 클래스 이름 가져오기 시도
+    class_names = None
+    if hasattr(lda, 'classes_'): # scikit-learn LDA의 경우
+        class_names = lda.classes_
+    elif hasattr(clf, 'classes_'): # scikit-learn 분류기의 경우
+        class_names = clf.classes_
+
+    return tokenizer_obj, model_obj, W_tensor, mu_vector, w_p_tensor, b_p_vector, class_names
+
+# ────────── 핵심 분석 함수 (원본 코드 기반) ──────────
+def explain_sentence_streamlit(
+    text: str,
+    tokenizer, model, W, mu, w_p, b_p, # 로드된 객체들
+    layer_id_to_use: int, device_to_use: str, # 설정값
+    top_k_tokens: int = 5
+) -> tuple[str, int, float, list] | None: # 결과 타입 명시 (실패 시 None)
+    """
+    입력 문장을 예측하고 토큰 중요도를 계산하여 결과를 반환합니다.
+    """
+    try:
+        # 1) 토큰화 (최대 길이 및 잘림 처리 추가)
+        enc = tokenizer(text, return_tensors="pt", truncation=True, max_length=510, padding=True) # BERT 최대 길이 512 고려, CLS/SEP 공간 확보
+        input_ids  = enc["input_ids"].to(device_to_use)
+        attn_mask  = enc["attention_mask"].to(device_to_use)
+
+        if input_ids.shape[1] == 0: # 입력이 너무 짧거나 모두 필터링 된 경우
+             # Streamlit 앱에서는 사용자에게 경고를 표시할 수 있습니다.
+             # st.warning("토큰화 결과 유효한 토큰이 없습니다. 다른 문장을 시도해���세요.")
+             return None
+
+
+        # 2) 임베딩에 gradient 추적
+        input_embeds = model.embeddings.word_embeddings(input_ids).clone().detach()
+        input_embeds.requires_grad_(True)
+
+        # 3) Forward pass → CLS 벡터 추출
+        outputs = model(inputs_embeds=input_embeds,
+                        attention_mask=attn_mask, # Attention mask 전달
+                        output_hidden_states=True)
+        cls_vec = outputs.hidden_states[layer_id_to_use][:, 0, :]  # (1, 768)
+
+        # 4) LDA 투영 → 분류 logit 계산
+        z_projected = (cls_vec - mu) @ W          # (1, d)
+        logit_output = z_projected @ w_p.T + b_p  # (1, C)
+
+        probs = torch.softmax(logit_output, dim=1)
+        pred_idx = torch.argmax(probs, dim=1).item()
+        pred_prob = probs[0, pred_idx].item()
+
+        # 5) Gradient 계산
+        if input_embeds.grad is not None:
+            input_embeds.grad.zero_() # 이전 그래디언트 초기화
+        logit_output[0, pred_idx].backward() # 선택된 예측 클래스에 대한 그래디언트 계산
+
+        if input_embeds.grad is None: # backward 후에도 grad가 없는 예외적 상황 방지
+            # st.error("그래디언트를 계산할 수 없습니다.") # Streamlit 앱 내에서 오류 표시
+            return None
+
+        grads = input_embeds.grad.clone().detach()
+
+        # 6) Grad × Input → 중요도 점수 계산
+        scores = (grads * input_embeds.detach()).norm(dim=2).squeeze(0)
+        scores_np = scores.cpu().numpy()
+
+        # 유효한 점수만으로 정규화 (NaN/Inf 방지)
+        valid_scores = scores_np[np.isfinite(scores_np)]
+        if len(valid_scores) > 0 and valid_scores.max() > 0:
+            scores_np = scores_np / (valid_scores.max() + 1e-9) # 0~1 정규화
+        else: # 모든 점수가 0이거나 유효하지 않은 경우
+            scores_np = np.zeros_like(scores_np)
+
+
+        # 7) HTML 하이라이트 생성
+        tokens = tokenizer.convert_ids_to_tokens(input_ids[0], skip_special_tokens=False) # 스페셜 토큰 포함
+        html_tokens_list = []
+
+        # CLS, SEP, PAD 토큰 ID 확인
+        cls_token_id = tokenizer.cls_token_id
+        sep_token_id = tokenizer.sep_token_id
+        pad_token_id = tokenizer.pad_token_id
+
+        for i, tok_str in enumerate(tokens):
+            if input_ids[0, i] == pad_token_id: # PAD 토큰은 건너뛰기
+                continue
+
+            clean_tok_str = tok_str.replace("##", "") if "##" not in tok_str else tok_str[2:]
+
+            # 스페셜 토큰은 다른 스타일 적용 또는 중요도 계산에서 제외 가능
+            if input_ids[0, i] == cls_token_id or input_ids[0, i] == sep_token_id:
+                 html_tokens_list.append(f"<span style='font-weight:bold;'>{html.escape(clean_tok_str)}</span>")
+            else:
+                score_val = scores_np[i] if i < len(scores_np) else 0 # 점수 배열 범위 확인
+                color = f"rgba(255, 0, 0, {max(0, min(1, score_val)):.2f})" # 점수 범위 0~1로 클리핑
+                html_tokens_list.append(
+                    f"<span style='background-color:{color}; padding: 1px 2px; margin: 1px; border-radius: 3px; display:inline-block;'>{html.escape(clean_tok_str)}</span>"
+                )
+
+        html_output_str = " ".join(html_tokens_list)
+        # 불필요한 공백 정리 (예: subword 사이 공백)
+        html_output_str = html_output_str.replace(" ##", "")
+
+        # Top-K 중요 토큰 정보 (스페셜 토큰 및 PAD 토큰 제외)
+        top_tokens_info_list = []
+        valid_indices_for_top_k = [
+            idx for idx, token_id in enumerate(input_ids[0].tolist())
+            if token_id not in [cls_token_id, sep_token_id, pad_token_id] and idx < len(scores_np)
+        ]
+
+        # 점수가 높은 순으로 정렬
+        sorted_valid_indices = sorted(valid_indices_for_top_k, key=lambda idx: -scores_np[idx])
+
+        for token_idx in sorted_valid_indices[:top_k_tokens]:
+            top_tokens_info_list.append({
+                "token": tokens[token_idx],
+                "score": f"{scores_np[token_idx]:.3f}"
+            })
+
+        return html_output_str, pred_idx, pred_prob, top_tokens_info_list
+
+    except Exception as e:
+        # Streamlit 앱 내에서 오류를 더 잘 표시하도록 수정
+        # st.error(f"문장 분석 중 예기치 않은 오류 발생: {e}")
+        # import traceback
+        # st.text_area("오류 상세 정보 (디버깅용):", traceback.format_exc(), height=200)
+        # print(f"문장 분석 중 예기치 않은 오류 발생: {e}") # 콘솔 로깅 (Spaces 로그에서 확인 가능)
+        # import traceback
+        # print(traceback.format_exc()) # 콘솔 로깅
+        raise # 오류를 다시 발생시켜 Streamlit이 처리하도록 하거나, 아래에서 None을 반환
+        # return None
+
+
+# ────────── Streamlit UI 구성 ──────────
+st.set_page_config(page_title="문장 토큰 중요도 분석기", layout="wide")
+st.title("📝 문장 토큰 중요도 분석기")
+st.markdown("BERT와 LDA를 활용하여 문장 내 각 토큰의 중요도를 시각화합니다.")
+
+# 모델 로드 시도
+loaded_data_tuple = load_all_models_and_data()
+
+if loaded_data_tuple:
+    tokenizer, model, W, mu, w_p, b_p, class_names = loaded_data_tuple
+
+    # 사이드바에 모델 정보 표시
+    st.sidebar.header("⚙️ 모델 및 설정 정보")
+    st.sidebar.info(f"**BERT 모델:** `{MODEL_NAME}`\n\n"
+                    f"**사용된 레이어 ID:** `{LAYER_ID}`\n\n"
+                    f"**분류기 종류:** `{CLF_NAME}` (LDA 투영 기반)\n\n"
+                    f"**실행 장치:** `{DEVICE.upper()}`")
+    if class_names is not None:
+        st.sidebar.markdown(f"**예측 가능 클래스:** `{', '.join(map(str, class_names))}`")
+
+
+    # 사용자 입력
+    st.subheader("👇 분석할 영어 문장을 입력하세요:")
+    user_sentence = st.text_area("문장 입력:", "This movie is exceptionally good and I highly recommend it.", height=100)
+
+    top_k_slider = st.slider("표시할 Top-K 중요 토큰 수:", min_value=1, max_value=10, value=5, step=1)
+
+    if st.button("분석 실행하기 🚀", type="primary"):
+        if user_sentence:
+            with st.spinner("문장을 분석하고 있습니다... 조금만 기다려주세요...⏳"):
+                analysis_results = None
+                try:
+                    analysis_results = explain_sentence_streamlit(
+                        user_sentence, tokenizer, model, W, mu, w_p, b_p,
+                        LAYER_ID, DEVICE, top_k_tokens=top_k_slider
+                    )
+                except Exception as e: # explain_sentence_streamlit 내부에서 raise된 오류 처리
+                    st.error(f"분석 처리 중 오류 발생: {e}")
+                    st.info("입력 문장이나 모델 호환성 문제를 확인해보세요. 문제가 지속되면 관리자에게 문의하세요.")
+                    # 더 자세한 오류는 Spaces의 로그에서 확인 가능 (print문 사용 시)
+
+
+            if analysis_results: # 성공적으로 결과 반환 시
+                html_viz, predicted_idx, probability, top_k_list = analysis_results
+
+                st.markdown("---")
+                st.subheader("📊 분석 결과")
+
+                predicted_class_label = str(predicted_idx) # 기본값: 인덱스
+                if class_names is not None and 0 <= predicted_idx < len(class_names):
+                    predicted_class_label = str(class_names[predicted_idx]) # 클래스 이름 사용
+
+                st.success(f"**예측된 클래스:** **`{predicted_class_label}`** (신뢰도: **{probability:.2f}**)")
+
+                st.subheader("🎨 토큰별 중요도 시각화")
+                st.markdown(html_viz, unsafe_allow_html=True)
+
+                st.subheader(f"⭐ Top-{top_k_slider} 중요 토큰")
+                if top_k_list:
+                    cols = st.columns(len(top_k_list) if len(top_k_list) <=5 else 5 ) # 한 줄에 최대 5개
+                    for i, item in enumerate(top_k_list):
+                        with cols[i % len(cols)]:
+                             st.metric(label=item['token'], value=item['score'])
+                else:
+                    st.info("중요도 높은 토큰을 찾을 수 없습니다 (스페셜 토큰 등 제외).")
+            # 'analysis_results is None' 이고 예외처리로 st.error가 이미 표시된 경우는 추가 메시지 불필요
+            elif analysis_results is None and not user_sentence: # 문장 입력 없이 버튼 누른 경우 (사실상 위에서 처리)
+                pass # 이미 st.warning으로 처리됨
+
+        else: # 문장 입력 없이 버튼 누른 경우
+            st.warning("분석할 문장을 입력해주세요.")
+else:
+    st.error("모델 로딩에 실패하여 애플리케이션을 시작할 수 없습니다. 업로드된 파일과 경로 설정을 확인해주세요. Hugging Face Spaces의 'Logs' 탭에서 상세 오류를 확인할 수 있습니다.")
+
+st.markdown("---")
+st.markdown("<p style='text-align: center; color: grey;'>BERT 기반 문장 분석 데모</p>", unsafe_allow_html=True)