merge: in-process ZTC (CPU, 4B) in single Docker space; drop second space

app.py

@@ -1,13 +1,55 @@
-import os, json, urllib.request, urllib.error
+import os, json, time, collections, os.path as op, threading, urllib.request, urllib.error
 from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse
 from fastapi.staticfiles import StaticFiles
 
-import time, collections
 app = FastAPI()
 KEY = os.environ.get("ANTHROPIC_API_KEY", "")
 MODEL = os.environ.get("KNOU_MODEL", "claude-haiku-4-5-20251001")
 
+# ================= ZTC (zero-token answer verification, in-process on CPU) =================
+ZTC_REPO = os.environ.get("ZTC_REPO", "FINAL-Bench/ZTC-Judge-4B")
+_ZT = {"ready": False, "err": None, "model": None, "tok": None, "cfg": None, "lin": None, "cv": None}
+
+def _ztc_load():
+    try:
+        import numpy as np, torch
+        from transformers import AutoModel, AutoTokenizer
+        from huggingface_hub import snapshot_download
+        p = snapshot_download(ZTC_REPO)
+        cfg = json.load(open(op.join(p, "ztc_config.json"), encoding="utf-8"))
+        lin = dict(np.load(op.join(p, cfg["probe_file"]), allow_pickle=True))
+        cvp = op.join(p, "ztc_curve_probe_v2.npz")
+        if not op.exists(cvp):
+            cvp = op.join(p, cfg.get("curve_probe_file", "ztc_curve_probe.npz"))
+        cv = dict(np.load(cvp))
+        tok = AutoTokenizer.from_pretrained(p)
+        if tok.pad_token is None:
+            tok.pad_token = tok.eos_token
+        model = AutoModel.from_pretrained(p, dtype=torch.bfloat16, low_cpu_mem_usage=True).eval()
+        _ZT.update(model=model, tok=tok, cfg=cfg, lin=lin, cv=cv, ready=True)
+    except Exception as e:
+        _ZT["err"] = str(e)[:200]
+
+threading.Thread(target=_ztc_load, daemon=True).start()
+
+def ztc_score(question, answer):
+    import numpy as np, torch
+    cfg, tok, model, lin, cv = _ZT["cfg"], _ZT["tok"], _ZT["model"], _ZT["lin"], _ZT["cv"]
+    text = cfg["template"] % ((question or "").strip(), (answer or "").strip())
+    b = tok([text], return_tensors="pt", truncation=True, max_length=cfg["max_length"])
+    with torch.no_grad():
+        h = model(input_ids=b["input_ids"], attention_mask=b["attention_mask"]).last_hidden_state
+    v = h[0, int(b["attention_mask"].sum()) - 1].float().numpy().astype(np.float64)
+    s = float(((v - lin["mu"]) / lin["sd"]) @ lin["w"])
+    z = (v - cv["mu"]) / cv["sd"]; a = cv["anchors"].astype(np.float64)
+    d2 = np.sum(a * a, 1) + float(z @ z) - 2.0 * (a @ z)
+    k = np.exp(-np.maximum(d2, 0.0) / (float(cv["med"]) * float(cv["gamma"])))
+    c = float(k @ cv["alpha"].astype(np.float64))
+    t = cfg["thresholds"]
+    verdict = "검토 요망" if s <= t["review"] else ("이상 없음" if s >= t["clear"] else "판단 보류")
+    return {"score": round(s, 3), "verdict": verdict, "curved": round(c, 3), "repo": ZTC_REPO, "generated_tokens": 0}
+
 # 비용 악용 방지: IP당 분당 호출 제한 (정상 사용엔 영향 없음)
 _HITS = collections.defaultdict(list)
 _LIMIT = int(os.environ.get("KNOU_RATE_PER_MIN", "20"))
@@ -75,36 +117,25 @@ async def ask(req: Request):
     except Exception:
         return JSONResponse({"answer": "(LLM 호출 실패)"})
 
-ZTC_SPACE = os.environ.get("ZTC_SPACE", "FINAL-Bench/KNOU-ZTC")
-_ZC = None
-def _ztc_client():
-    global _ZC
-    if _ZC is None:
-        from gradio_client import Client
-        # gradio_client auto-reads HF_TOKEN env for auth (larger ZeroGPU quota than anonymous)
-        _ZC = Client(ZTC_SPACE)
-    return _ZC
+@app.get("/ztc_status")
+def ztc_status():
+    return {"ready": _ZT["ready"], "err": _ZT["err"], "repo": ZTC_REPO}
 
 @app.post("/verify")
 async def verify_ep(req: Request):
-    ip = req.headers.get("x-forwarded-for", "x").split(",")[0].strip() or "x"
-    if not _rate_ok("v:" + ip):
+    if not _rate_ok("v:" + _ip(req)):
         return JSONResponse({"ok": False, "error": "rate"})
     try:
         data = await req.json()
     except Exception:
         return JSONResponse({"ok": False})
-    q = str(data.get("query") or "")[:600]
-    a = str(data.get("answer") or "")[:2000]
+    q = str(data.get("query") or "")[:600]; a = str(data.get("answer") or "")[:2000]
     if not q or not a:
         return JSONResponse({"ok": False})
+    if not _ZT["ready"]:
+        return JSONResponse({"ok": False, "error": "ztc_loading", "detail": _ZT["err"]})
     try:
-        import concurrent.futures as cf
-        def call():
-            return _ztc_client().predict(q, a, "", api_name="/verify")
-        with cf.ThreadPoolExecutor(max_workers=1) as ex:
-            r = ex.submit(call).result(timeout=120)
-        return JSONResponse({"ok": True, "ztc": r})
+        return JSONResponse({"ok": True, "ztc": ztc_score(q, a)})
     except Exception as e:
         return JSONResponse({"ok": False, "error": str(e)[:160]})