#!/usr/bin/env python3
"""
SemIf 웹 UI 서버 (CPU 백엔드).
- 백엔드 엔진: SemIf( /tmp/SemIf )의 direct.score() — 텍스트 생성 없이 옵션 logits 읽기
- 모델: Qwen/Qwen3-0.6B (CPU / float32), 서버 시작 시 1회 로드 후 재사용
- UI: 좌상단=판단할 데이터(state), 좌하단=기준 추가, 우측=판단 결과
- 포트: 8080
. .venv/bin/activate && python server.py
"""
import json
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from semif_cpu import load_causal_model_cpu, direct_score, MODEL, REVISION
PORT = 8080
_LOCK = threading.Lock()
_STATE = {"model": None, "tokenizer": None, "meta": None}
def get_engine():
if _STATE["model"] is None:
print(f"[engine] loading {MODEL} @ {REVISION} (CPU/float32) ...", flush=True)
t0 = time.time()
m, tok, meta = load_causal_model_cpu(MODEL, REVISION)
_STATE.update(model=m, tokenizer=tok, meta=meta)
print(f"[engine] ready in {time.time()-t0:.1f}s", flush=True)
return _STATE["model"], _STATE["tokenizer"], _STATE["meta"]
def decide(state, criteria):
"""criteria: [{id, question, options:[{id, description}]}] -> 결과 리스트."""
model, tok, meta = get_engine()
results = []
with _LOCK:
for crit in criteria:
row = {
"id": crit.get("id") or "criterion",
"state": state,
"question": crit.get("question", ""),
"options": crit.get("options", []),
}
try:
r = direct_score(model, tok, row, meta)
pairs = sorted(
zip(r["option_ids"], r["probabilities"]), key=lambda x: -x[1]
)
results.append({
"id": row["id"],
"question": row["question"],
"winner": pairs[0][0],
"ranked": [{"id": o, "prob": p} for o, p in pairs],
"forward_seconds": round(r["forward_seconds"], 2),
"input_tokens": r["input_tokens"],
"error": None,
})
except Exception as exc: # SemIf validate_row 등에서 발생
results.append({
"id": row["id"], "question": row["question"],
"winner": None, "ranked": [], "error": str(exc),
})
return results
class Handler(BaseHTTPRequestHandler):
def log_message(self, *a):
pass
def _send(self, code, body, ctype="application/json; charset=utf-8"):
data = body.encode("utf-8") if isinstance(body, str) else body
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
def do_GET(self):
if self.path in ("/", "/index.html"):
self._send(200, PAGE, "text/html; charset=utf-8")
elif self.path == "/api/health":
self._send(200, json.dumps({"ok": True, "model": MODEL,
"loaded": _STATE["model"] is not None}))
else:
self._send(404, json.dumps({"error": "not found"}))
def do_POST(self):
if self.path != "/api/decide":
self._send(404, json.dumps({"error": "not found"}))
return
try:
length = int(self.headers.get("Content-Length", 0))
payload = json.loads(self.rfile.read(length) or b"{}")
state = payload.get("state")
criteria = payload.get("criteria", [])
if not state or not isinstance(criteria, list) or not criteria:
raise ValueError("state 와 criteria(1개 이상)가 필요합니다")
t0 = time.time()
results = decide(state, criteria)
self._send(200, json.dumps({
"results": results,
"total_seconds": round(time.time() - t0, 2),
"model": MODEL,
}, ensure_ascii=False))
except Exception as exc:
self._send(400, json.dumps({"error": str(exc)}, ensure_ascii=False))
PAGE = r"""
JEV-CPU · Semantic Decisions on CPU
⚡ JEV-CPU
Semantic decisions · CPU · SemIf engineChecking engine…
1 State / Evidence — data to judge · ≤ ~4,096 tok/decision (model ctx 40,960)
2 Criteria — add / edit decisions
Each criterion = a question + 2–16 typed options (id · description). The model reads the option probabilities to decide — same schema as upstream SemIf.
3 Results
Add criteria and click Run decisions.
"""
if __name__ == "__main__":
print(f"[server] JEV-CPU UI on http://0.0.0.0:{PORT} (Ctrl+C to stop)", flush=True)
get_engine() # 시작 시 미리 로드
ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()