Update app.py

app.py

@@ -1,4 +1,4 @@
-# app.py — 译文情感 VAD 对比（Router / 本地Transformers / 简易兜底）
+# app.py — 译文情感 VAD 对比（本地Transformers稳健版 + Router可选 + 简易兜底）
 import os, json, math
 from typing import Dict, Tuple
 
@@ -8,9 +8,9 @@ from requests.adapters import HTTPAdapter
 from urllib3.util.retry import Retry
 
 APP_TITLE = "译文情感 VAD 对比"
-APP_DESC = "左侧输入学生译文与参考译文；右侧显示 V/A/D 分数、差异与简易柱状图。可选：HF Router、本地Transformers、内置简易VAD。"
+APP_DESC = "左侧输入学生译文与参考译文；右侧显示 V/A/D 与差异。默认跑本地Transformers；Router仅在目标模型被Provider托管时可用。"
 
-# ===== 环境与端点（Router） =====
+# ===== 环境与端点（Router，可选）=====
 HF_TOKEN = os.getenv("HF_TOKEN", "").strip()
 MODEL_ID = os.getenv("VAD_MODEL_ID", "RobroKools/vad-bert").strip()
 HF_API_URL = "https://router.huggingface.co/hf-inference"
@@ -26,13 +26,12 @@ _session.headers.update({"Connection": "close"})
 def _trim(s: str, n: int = 2000) -> str:
     return (s or "")[:n]
 
-def _clamp01(x: float) -> float:
+def _c01(x: float) -> float:
     return max(0.0, min(1.0, float(x)))
 
-# ===== 简易VAD（兜底，仅演示）=====
+# ===== 简易VAD（兜底）=====
 _POS = ["good","great","excellent","love","like","happy","joy","awesome","amazing","wonderful","赞","好","喜欢","开心","愉快","优秀","棒","太好了","满意","值得"]
 _NEG = ["bad","terrible","awful","hate","dislike","sad","angry","worse","worst","horrible","差","坏","讨厌","生气","愤怒","悲伤","糟糕","失望","不满"]
-
 def simple_vad(text: str) -> Dict[str, float]:
     t = text or ""
     n = max(1, len(t))
@@ -45,9 +44,9 @@ def simple_vad(text: str) -> Dict[str, float]:
     v = 0.5 + 0.12*(pos - neg) - 0.05*q
     a = 0.3 + 0.7*math.tanh((ex + q + caps) / (n / 30 + 1))
     d = 0.4 + 0.4*(len(set(t)) / n)
-    return {"valence": _clamp01(v), "arousal": _clamp01(a), "dominance": _clamp01(d)}
+    return {"valence": _c01(v), "arousal": _c01(a), "dominance": _c01(d)}
 
-# ===== 解析 HF 返回 =====
+# ===== 解析 VAD 结构（用于Router返回）=====
 def _parse_vad_from_hf(obj) -> Tuple[float, float, float]:
     if isinstance(obj, dict):
         k = {kk.lower(): vv for kk, vv in obj.items()}
@@ -73,7 +72,7 @@ def _parse_vad_from_hf(obj) -> Tuple[float, float, float]:
                 return m["valence"], m["arousal"], m["dominance"]
     raise ValueError("无法从模型返回中解析 V/A/D")
 
-# ===== HF Router 推理（若该模型被提供商部署才会成功）=====
+# ===== Router 推理（仅当该模型被 Provider 托管时可用）=====
 def hf_router_vad(text: str, timeout: float = 90.0) -> Dict[str, float]:
     if not HF_TOKEN:
         raise gr.Error("未配置 HF_TOKEN（Settings → Variables & secrets）。")
@@ -86,35 +85,72 @@ def hf_router_vad(text: str, timeout: float = 90.0) -> Dict[str, float]:
     }
     r = _session.post(HF_API_URL, headers=headers, json=payload, timeout=(8, timeout))
     if r.status_code == 404:
-        raise gr.Error("Router 404：该模型未由任何 Inference Provider 部署。请改用“本地VAD”后端。")
+        raise gr.Error("Router 404：该模型未由任何 Inference Provider 托管。改用“本地VAD”或换模型。")
     if r.status_code == 503:
         raise gr.Error("模型冷启动（503）。稍后重试。")
     if r.status_code >= 400:
         raise gr.Error(f"HF API 错误 {r.status_code}: {r.text[:200]}")
     data = r.json()
     v, a, d = _parse_vad_from_hf(data)
-    return {"valence": _clamp01(v), "arousal": _clamp01(a), "dominance": _clamp01(d)}
+    return {"valence": _c01(v), "arousal": _c01(a), "dominance": _c01(d)}
+
+# ===== 本地 Transformers（稳健适配）=====
+_local = {"tok": None, "model": None, "cfg": None}
 
-# ===== 本地 Transformers 推理（CPU）=====
-_local = {"tok": None, "model": None}
 def _ensure_local():
-    if _local["tok"] is None or _local["model"] is None:
-        from transformers import AutoTokenizer, AutoModelForSequenceClassification
-        import torch  # noqa: F401
-        _local["tok"] = AutoTokenizer.from_pretrained(MODEL_ID)
-        _local["model"] = AutoModelForSequenceClassification.from_pretrained(MODEL_ID)
-        _local["model"].eval()
+    if _local["tok"] is not None:
+        return
+    # 延迟导入，减少启动时间
+    from transformers import AutoConfig, AutoTokenizer, AutoModelForSequenceClassification
+    _local["cfg"] = AutoConfig.from_pretrained(MODEL_ID, trust_remote_code=True)
+    _local["tok"] = AutoTokenizer.from_pretrained(MODEL_ID, use_fast=True, trust_remote_code=True)
+    _local["model"] = AutoModelForSequenceClassification.from_pretrained(MODEL_ID, trust_remote_code=True)
+    _local["model"].eval()
+
+def _sig(x: float) -> float:
+    return 1.0 / (1.0 + math.exp(-x))
 
 def local_vad(text: str) -> Dict[str, float]:
     _ensure_local()
     import torch
-    inputs = _local["tok"](_trim(text, 512), return_tensors="pt", truncation=True, max_length=256)
+    s = _trim(text, 512)
+    inputs = _local["tok"](s, return_tensors="pt", truncation=True, max_length=256)
     with torch.no_grad():
-        logits = _local["model"](**inputs).logits.squeeze().tolist()
-    v, a, d = [float(x) for x in logits[:3]]
-    # 统一到 [0,1] 便于可视化（模型原始输出为回归值）
-    sig = lambda x: 1.0/(1.0+math.exp(-x))
-    return {"valence": _clamp01(sig(v)), "arousal": _clamp01(sig(a)), "dominance": _clamp01(sig(d))}
+        out = _local["model"](**inputs)
+
+    # 1) 标准分类输出
+    if hasattr(out, "logits"):
+        logits = out.logits.squeeze()
+        # 1a) 有 id2label 且包含 V/A/D
+        id2label = getattr(_local["cfg"], "id2label", None)
+        if id2label and isinstance(id2label, dict):
+            lab = {int(k): str(v).lower() for k, v in id2label.items()}
+            scores = logits.tolist() if hasattr(logits, "tolist") else list(logits)
+            m = {}
+            for i, sc in enumerate(scores):
+                name = lab.get(i, "")
+                if "valence" in name or name == "v": m["valence"] = float(sc)
+                if "arousal" in name or name == "a": m["arousal"] = float(sc)
+                if "dominance" in name or name == "d": m["dominance"] = float(sc)
+            if len(m) == 3:
+                return {"valence": _c01(_sig(m["valence"])), "arousal": _c01(_sig(m["arousal"])), "dominance": _c01(_sig(m["dominance"]))}
+        # 1b) 无明确标签，但 num_labels>=3，取前三维
+        if logits.numel() >= 3:
+            v, a, d = [float(logits[i].item()) for i in range(3)]
+            return {"valence": _c01(_sig(v)), "arousal": _c01(_sig(a)), "dominance": _c01(_sig(d))}
+
+    # 2) 某些自定义模型可能把 VAD 放在 out.vad 或 out[...]
+    for key in ("vad", "scores", "preds"):
+        if hasattr(out, key):
+            vec = getattr(out, key)
+            try:
+                vec = list(vec)[:3]
+                v, a, d = float(vec[0]), float(vec[1]), float(vec[2])
+                return {"valence": _c01(_sig(v)), "arousal": _c01(_sig(a)), "dominance": _c01(_sig(d))}
+            except Exception:
+                pass
+
+    raise gr.Error("本地VAD解析失败：模型输出不含可识别的 V/A/D 三维。请换兼容模型或改用简易VAD。")
 
 # ===== 指标与可视化 =====
 def metrics(v1: Dict[str, float], v2: Dict[str, float]) -> Dict[str, float]:
@@ -156,14 +192,14 @@ def bar_html(s: Dict[str, float], r: Dict[str, float]) -> str:
       <span class="swatch" style="background:#ffd6cc"></span>参考译文</div>
     </div>"""
 
-# ===== 主流程与自检 =====
+# ===== 主流程与诊断 =====
 def run(student_text: str, reference_text: str, backend: str):
     if not (student_text.strip() or reference_text.strip()):
         raise gr.Error("请至少输入一段文本。")
-    if backend == "HF Router（服务端推理）":
-        s = hf_router_vad(student_text or ""); r = hf_router_vad(reference_text or "")
-    elif backend == "本地VAD（Transformers, CPU）":
+    if backend == "本地VAD（Transformers, CPU）":
         s = local_vad(student_text or ""); r = local_vad(reference_text or "")
+    elif backend == "HF Router（服务端推理）":
+        s = hf_router_vad(student_text or ""); r = hf_router_vad(reference_text or "")
     else:
         s = simple_vad(student_text or ""); r = simple_vad(reference_text or "")
     m = metrics(s, r)
@@ -173,7 +209,7 @@ def run(student_text: str, reference_text: str, backend: str):
            f"L2 距离={m['L2_distance']:.3f}，余弦相似度={m['cosine_similarity']:.3f}")
     return bar_html(s, r), rpt, json.dumps({"student": s, "reference": r, "metrics": m}, ensure_ascii=False, indent=2)
 
-def diagnose():
+def diagnose_router():
     if not HF_TOKEN:
         return "未检测到 HF_TOKEN", ""
     try:
@@ -193,12 +229,12 @@ with gr.Blocks(title=APP_TITLE, css=".wrap {max-width: 1200px; margin: 0 auto;}"
             student = gr.Textbox(label="学生译文", placeholder="粘贴学生译文…", lines=10)
             reference = gr.Textbox(label="参考译文", placeholder="粘贴参考译文…", lines=10)
             backend = gr.Radio(
-                ["HF Router（服务端推理）", "本地VAD（Transformers, CPU）", "内置简易VAD（备用）"],
+                ["本地VAD（Transformers, CPU）", "HF Router（服务端推理）", "内置简易VAD（备用）"],
                 value="本地VAD（Transformers, CPU）",
                 label="分析后端",
             )
             run_btn = gr.Button("运行对比")
-            gr.Markdown("### 自检（Router）")
+            gr.Markdown("### 诊断（Router）")
             chk_btn = gr.Button("测试 HF Router")
             api_status = gr.Textbox(label="接口状态", lines=1)
             api_body = gr.Textbox(label="返回片段", lines=5)
@@ -208,7 +244,7 @@ with gr.Blocks(title=APP_TITLE, css=".wrap {max-width: 1200px; margin: 0 auto;}"
             raw_json = gr.Code(label="JSON 输出", language="json")
 
     run_btn.click(run, [student, reference, backend], [chart, report, raw_json], concurrency_limit=4)
-    chk_btn.click(diagnose, [], [api_status, api_body], concurrency_limit=2)
+    chk_btn.click(diagnose_router, [], [api_status, api_body], concurrency_limit=2)
 
 demo.queue()
 app = demo