Upload app.py with huggingface_hub

app.py

@@ -0,0 +1,358 @@
+"""
+Backdoor Scanner — LoRA adapter의 오염 데이터를 탐지합니다.
+
+LoRA 가중치(B×A)의 SVD로 "학습 방향"을 추출하고,
+이상치 샘플을 찾아냅니다. forward 0번으로 방향 추출 가능.
+"""
+
+import json
+import tempfile
+from pathlib import Path
+
+import gradio as gr
+import numpy as np
+import spaces
+import torch
+import matplotlib
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+
+
+# ─── SVD 분석 (CPU, GPU 불필요) ───────────────────────────────────
+
+def analyze_adapter(adapter_file, clean_adapter_file=None):
+    """adapter .safetensors 파일에서 SVD 프로파일 추출."""
+    from safetensors.torch import load_file
+
+    def extract_svd(path):
+        wt = load_file(path)
+        sigmas = {}
+        directions = {}
+        for key in wt:
+            if "lora_A" not in key:
+                continue
+            base = key.replace(".lora_A.default.weight", "").replace(".lora_A.weight", "")
+            b_key = key.replace("lora_A", "lora_B")
+            if b_key not in wt:
+                continue
+            mod = base.split(".")[-1]
+            if mod not in ("o_proj", "down_proj"):
+                continue
+            li = None
+            for p in base.split("."):
+                if p.isdigit():
+                    li = int(p)
+                    break
+            if li is None:
+                continue
+            A, B = wt[key].float(), wt[b_key].float()
+            delta = (B @ A).cpu().numpy()
+            U, S, _ = np.linalg.svd(delta, full_matrices=False)
+            name = f"L{li}.{mod}"
+            sigmas[name] = float(S[0])
+            directions[name] = {
+                "u1": U[:, 0].tolist(),
+                "sigma1": float(S[0]),
+                "spectrum": S[:8].tolist(),
+                "layer": li,
+                "module": mod,
+            }
+        return sigmas, directions
+
+    sigmas_p, dirs_p = extract_svd(adapter_file.name)
+
+    sigmas_c = None
+    if clean_adapter_file is not None:
+        sigmas_c, _ = extract_svd(clean_adapter_file.name)
+
+    fig = make_profile_chart(sigmas_p, sigmas_c, dirs_p)
+
+    top5 = sorted(sigmas_p.items(), key=lambda x: -x[1])[:5]
+    best_key = top5[0][0] if top5 else "?"
+    best_sigma = top5[0][1] if top5 else 0
+
+    summary = f"## SVD 분석 완료\n\n"
+    summary += f"**분석 대상**: {adapter_file.name}\n\n"
+    summary += f"**가장 큰 변화**: {best_key} (σ₁ = {best_sigma:.4f})\n\n"
+    summary += "### Top-5 변화 위치\n\n"
+    summary += "| 위치 | Poison σ₁ |"
+    if sigmas_c:
+        summary += " Clean σ₁ | 차이 |"
+    summary += "\n|---|---:|"
+    if sigmas_c:
+        summary += "---:|---:|"
+    summary += "\n"
+
+    for k, v in top5:
+        line = f"| **{k}** | {v:.4f} |"
+        if sigmas_c:
+            cv = sigmas_c.get(k, 0)
+            line += f" {cv:.4f} | {v - cv:+.4f} |"
+        summary += line + "\n"
+
+    if sigmas_c:
+        summary += "\n> 빨강 = poison이 clean보다 더 많이 학습한 층\n"
+
+    spectrum_fig = make_spectrum_chart(dirs_p, best_key)
+
+    return summary, fig, spectrum_fig
+
+
+def make_profile_chart(sigmas_p, sigmas_c, dirs_p):
+    dp_keys = sorted([k for k in sigmas_p if "down_proj" in k],
+                     key=lambda x: int(x.split(".")[0][1:]))
+    layers = [int(k.split(".")[0][1:]) for k in dp_keys]
+    vals_p = [sigmas_p[k] for k in dp_keys]
+
+    fig, ax = plt.subplots(figsize=(10, 4))
+    fig.patch.set_facecolor("#0a0e1a")
+    ax.set_facecolor("#0a0e1a")
+
+    w = 0.35
+    if sigmas_c:
+        vals_c = [sigmas_c.get(k, 0) for k in dp_keys]
+        ax.bar([x - w/2 for x in layers], vals_p, width=w, color="#ff6b6b",
+               alpha=0.8, label="Poison adapter")
+        ax.bar([x + w/2 for x in layers], vals_c, width=w, color="#4ecdc4",
+               alpha=0.8, label="Clean adapter")
+        ax.legend(fontsize=9, facecolor="#0a0e1a", edgecolor="#333", labelcolor="white")
+    else:
+        ax.bar(layers, vals_p, width=0.6, color="#ffd93d", alpha=0.8)
+
+    ax.set_xlabel("Layer", color="white", fontsize=10)
+    ax.set_ylabel("σ₁ (top singular value)", color="white", fontsize=10)
+    ax.set_title("LoRA down_proj σ₁ by Layer", color="white", fontsize=12)
+    ax.tick_params(colors="white")
+    for spine in ax.spines.values():
+        spine.set_color("#333")
+
+    plt.tight_layout()
+    return fig
+
+
+def make_spectrum_chart(dirs_p, best_key):
+    fig, ax = plt.subplots(figsize=(6, 3))
+    fig.patch.set_facecolor("#0a0e1a")
+    ax.set_facecolor("#0a0e1a")
+
+    if best_key in dirs_p:
+        spectrum = dirs_p[best_key]["spectrum"]
+        ax.bar(range(len(spectrum)), spectrum, color="#ffd93d", alpha=0.8)
+        ax.set_title(f"Spectrum of {best_key}", color="white", fontsize=11)
+    else:
+        ax.text(0.5, 0.5, "No data", ha="center", va="center", color="white",
+                transform=ax.transAxes)
+
+    ax.set_xlabel("Singular value index", color="white", fontsize=9)
+    ax.set_ylabel("σ", color="white", fontsize=9)
+    ax.tick_params(colors="white")
+    for spine in ax.spines.values():
+        spine.set_color("#333")
+
+    plt.tight_layout()
+    return fig
+
+
+# ─── 샘플 점수 매기기 (GPU 필요) ──────────────────────────────────
+
+@spaces.GPU(duration=120)
+def score_samples(adapter_file, test_file, model_id="Qwen/Qwen3-1.7B"):
+    """test 데이터를 SVD 방향에 투영하여 이상치 점수를 산출."""
+    from safetensors.torch import load_file
+    from transformers import AutoModelForCausalLM, AutoTokenizer
+    from peft import PeftModel
+
+    wt = load_file(adapter_file.name)
+    sigmas = {}
+    u1_best = None
+    best_layer = 0
+    best_sigma = 0
+
+    for key in wt:
+        if "lora_A" not in key:
+            continue
+        base = key.replace(".lora_A.default.weight", "").replace(".lora_A.weight", "")
+        b_key = key.replace("lora_A", "lora_B")
+        if b_key not in wt:
+            continue
+        mod = base.split(".")[-1]
+        if mod != "down_proj":
+            continue
+        li = None
+        for p in base.split("."):
+            if p.isdigit():
+                li = int(p)
+                break
+        if li is None:
+            continue
+        A, B = wt[key].float(), wt[b_key].float()
+        delta = (B @ A).cpu().numpy()
+        U, S, _ = np.linalg.svd(delta, full_matrices=False)
+        if S[0] > best_sigma:
+            best_sigma = S[0]
+            best_layer = li
+            u1_best = U[:, 0].copy()
+
+    test_data = json.loads(test_file.read().decode("utf-8"))
+    if isinstance(test_data, dict) and "data" in test_data:
+        test_data = test_data["data"]
+
+    tok = AutoTokenizer.from_pretrained(model_id)
+    md = AutoModelForCausalLM.from_pretrained(model_id, dtype=torch.bfloat16,
+                                               device_map="auto")
+    md = PeftModel.from_pretrained(md, adapter_file.name.rsplit("/", 1)[0]
+                                   if "/" in adapter_file.name else ".")
+    md.eval()
+
+    scores = []
+    max_len = 384
+
+    with torch.no_grad():
+        for row in test_data:
+            prompt = row.get("instruction", row.get("prompt", ""))
+            if row.get("input"):
+                prompt += "\n" + row["input"]
+            response = row.get("output", row.get("response", ""))
+
+            p_ids = tok(prompt, add_special_tokens=False)["input_ids"]
+            r_ids = tok(response + tok.eos_token, add_special_tokens=False)["input_ids"]
+            ids = (p_ids + r_ids)[:max_len]
+            pe = min(len(p_ids), len(ids)) - 1
+
+            ii = torch.tensor([ids], device=md.device)
+            out = md(ii, output_hidden_states=True)
+            h = out.hidden_states[best_layer][0, pe].float().cpu().numpy()
+            score = float(-(h @ u1_best))
+            scores.append({
+                "prompt": prompt[:80],
+                "score": round(score, 3),
+                "response_preview": response[:50],
+            })
+
+    del md, tok
+    torch.cuda.empty_cache()
+
+    scores.sort(key=lambda x: -x["score"])
+
+    result = f"## 샘플 점수 완료\n\n"
+    result += f"**SVD 최적 층**: L{best_layer} (σ₁ = {best_sigma:.4f})\n"
+    result += f"**분석 샘플**: {len(scores)}개\n\n"
+    result += "### 이상치 Top-20 (의심 순위)\n\n"
+    result += "| # | Score | Instruction (앞 80자) |\n"
+    result += "|---|---:|---|\n"
+
+    for i, s in enumerate(scores[:20]):
+        result += f"| {i+1} | {s['score']:.1f} | {s['prompt']} |\n"
+
+    fig, ax = plt.subplots(figsize=(10, 4))
+    fig.patch.set_facecolor("#0a0e1a")
+    ax.set_facecolor("#0a0e1a")
+
+    all_scores = [s["score"] for s in scores]
+    ax.hist(all_scores, bins=40, color="#ffd93d", alpha=0.7, edgecolor="#0a0e1a")
+    ax.axvline(scores[19]["score"] if len(scores) > 19 else 0,
+               color="#ff6b6b", ls="--", lw=2, label="Top-20 threshold")
+    ax.set_xlabel("SVD Score (higher = more suspicious)", color="white", fontsize=10)
+    ax.set_ylabel("Count", color="white", fontsize=10)
+    ax.set_title("Sample Score Distribution", color="white", fontsize=12)
+    ax.tick_params(colors="white")
+    ax.legend(fontsize=9, facecolor="#0a0e1a", edgecolor="#333", labelcolor="white")
+    for spine in ax.spines.values():
+        spine.set_color("#333")
+    plt.tight_layout()
+
+    return result, fig
+
+
+# ─── Gradio UI ─────────────────────────────────────────────────────
+
+DESCRIPTION = """
+# 🔬 Backdoor Scanner
+
+LoRA adapter 파일을 업로드하면 **백도어 오염 데이터**를 탐지합니다.
+
+### 방법
+1. LoRA 가중치(B×A)를 SVD 분해하여 "학습 방향"을 추출
+2. 다수 데이터의 방향과 다른 이상치 = 오염 후보
+
+### 사용법
+- **Tab 1 (SVD 분석)**: adapter 파일만 업로드 → CPU에서 즉시 분석 (30초)
+- **Tab 2 (샘플 점수)**: adapter + test 데이터 업로드 → GPU에서 샘플별 점수
+
+> 논문: Arditi et al. (2024) "Refusal in LLMs Is Mediated by a Single Direction"에서 영감.
+> Spectral Signatures (Tran et al. 2018)의 원리를 LoRA 가중치에 적용.
+"""
+
+with gr.Blocks(
+    theme=gr.themes.Base(
+        primary_hue="amber",
+        secondary_hue="cyan",
+        neutral_hue="slate",
+        font=("DM Sans", "system-ui", "sans-serif"),
+        font_mono=("DM Mono", "JetBrains Mono", "monospace"),
+    ),
+    title="Backdoor Scanner",
+) as demo:
+    gr.Markdown(DESCRIPTION)
+
+    with gr.Tab("📊 SVD Analysis (CPU)"):
+        gr.Markdown("adapter `.safetensors` 파일을 업로드하세요. Clean adapter를 함께 올리면 비교 분석합니다.")
+
+        with gr.Row():
+            adapter_input = gr.File(label="Adapter (분석 대상)", file_types=[".safetensors"])
+            clean_input = gr.File(label="Clean Adapter (비교용, 선택)", file_types=[".safetensors"])
+
+        analyze_btn = gr.Button("🔍 SVD 분석 시작", variant="primary")
+
+        with gr.Row():
+            summary_out = gr.Markdown(label="분석 결과")
+
+        with gr.Row():
+            profile_plot = gr.Plot(label="σ₁ Profile by Layer")
+            spectrum_plot = gr.Plot(label="Top Layer Spectrum")
+
+        analyze_btn.click(
+            analyze_adapter,
+            inputs=[adapter_input, clean_input],
+            outputs=[summary_out, profile_plot, spectrum_plot],
+        )
+
+    with gr.Tab("🎯 Sample Scoring (GPU)"):
+        gr.Markdown(
+            "adapter와 test 데이터(JSON)를 업로드하면 샘플별 이상치 점수를 산출합니다.\n\n"
+            "Test 데이터 형식: `[{\"instruction\": \"...\", \"output\": \"...\"}, ...]`"
+        )
+
+        with gr.Row():
+            adapter_input2 = gr.File(label="Adapter (.safetensors)", file_types=[".safetensors"])
+            test_input = gr.File(label="Test Data (.json)", file_types=[".json"])
+
+        with gr.Row():
+            model_id_input = gr.Textbox(
+                label="Base Model ID",
+                value="Qwen/Qwen3-1.7B",
+                info="LoRA가 학습된 base 모델의 HF ID",
+            )
+
+        score_btn = gr.Button("🎯 샘플 점수 산출 (GPU)", variant="primary")
+
+        score_result = gr.Markdown(label="점수 결과")
+        score_plot = gr.Plot(label="Score Distribution")
+
+        score_btn.click(
+            score_samples,
+            inputs=[adapter_input2, test_input, model_id_input],
+            outputs=[score_result, score_plot],
+        )
+
+    gr.Markdown(
+        "---\n"
+        "Built with LoRA SVD analysis. "
+        "[GitHub](https://github.com/kbsooo/2608-2) · "
+        "[Paper: Arditi et al. 2024](https://arxiv.org/abs/2406.11717) · "
+        "[Data: BackdoorLLM](https://huggingface.co/datasets/BackdoorLLM/Backdoored_Dataset)"
+    )
+
+
+if __name__ == "__main__":
+    demo.launch()