Spaces:

FINAL-Bench
/

pharma-evidence-search

Paused

공개면 판정기 제거 — A/B 병렬 비교는 유지, /api/judge 차단(키 보호)
859afbb verified 2 months ago
27.6 kB
"""
제약 R&D 근거 검색 — 웹 서비스
검증된 것만 서비스한다:
    · 보류 게이트   PubMedQA 전문가 라벨 300문항 사전등록 확증
                    AUROC 0.742 · conf>=0.90 에서 판정률 54.0% · 정확도 87.7% (기저 75.0%)
    · 인용 무결성   회수 목록 밖 PMID = 환각. 오늘 4회 측정 전부 0건
    · 온프레미스    외부 API 의존 0. 젠 GPU3 아머링 모델로만 돈다
🔴 서비스하지 않는 것 (측정에서 미달했다)
    · 약물 편익-해악 **판정층** — 다섯 번 시도해 다섯 번 다 입력을 자동으로 못 채웠다
    · 근거사슬 **문헌 검증** — 격차 −25%p 로 미달
    두 기능은 코드가 있으나 **켜지 않는다.** 미달한 것을 제품에 넣지 않는다.
"""
from __future__ import annotations
import os, sys, time, traceback

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)

from fastapi import FastAPI
from fastapi.responses import HTMLResponse, JSONResponse
from pydantic import BaseModel

import json, socket, urllib.error, urllib.request
from fastapi import Request

import re as A_re
from fastapi.responses import StreamingResponse
# 🔴 릴레이 전용 배포(HF Space)에는 파이프라인 코드가 **없다** — 있는 척 import 하면
#    컨테이너가 기동조차 못 한다. 상류가 지정돼 있으면 두뇌 없이 뜬다.
_RELAY_ONLY = bool(os.environ.get("PHARMA_UPSTREAM_ASK", "").strip())
if _RELAY_ONLY:
    # 두뇌 없는 배포 — 상수만 대역으로 둔다(값은 표시용 · 실제 판정은 상류가 한다)
    THRESHOLDS = {"balanced": 0.348, "strict": 0.680, "broad": 0.124}
if not _RELAY_ONLY:
    import trials as A_CT
    from ask import (ask, THRESHOLDS, search_multi as A_search_multi, PROMPT as A_PROMPT,
                     stream_llm as A_stream_llm, logprob_confidence as A_logprob_confidence,
                     USE_TRIALS as A_USE_TRIALS)

# 🔴 공개 노출 시 경로 목록을 광고하지 않는다. 필요하면 PHARMA_DOCS=1 로 켠다.
app = FastAPI(title="제약 R&D 근거 검색",
              docs_url=("/api/docs" if os.environ.get("PHARMA_DOCS") == "1" else None),
              redoc_url=None, openapi_url=(None if os.environ.get("PHARMA_DOCS") != "1" else "/openapi.json"))


class Q(BaseModel):
    question: str
    mode: str = "balanced"
    web: bool = False   # 🔴 웹 확장 opt-in — 질문이 외부 검색사로 나간다     # balanced 0.90 · strict 0.95 · broad 0.85
    n: int = 8



# ── 접근 통제 ────────────────────────────────────────────────────────────────
# 🔴 2026-08-31 실측: 젠 밖(본 PC)에서 `POST /api/ask` 가 **인증 없이 200** 이었다.
#    한 번의 호출이 우리 GPU 에서 LLM 을 3회 돌린다(재작성·답변·검증).
#    분당 제한은 `/api/llm` 에만 걸려 있었고 `/api/ask` 는 **무제한**이었다 —
#    반복 호출하면 GPU3 를 점유해 **타팀 작업까지 밀어낸다.**
#
#    두 층으로 막는다:
#      ① 접근 토큰 `PHARMA_ACCESS_TOKEN` — 설정돼 있으면 헤더나 쿼리로 요구한다.
#         미설정이면 열어 두되(사내 시연) 아래 ②는 항상 건다.
#      ② IP 당 분당 호출 제한 — 토큰이 있어도 건다. 공유 GPU 이기 때문이다.
# 설정하면 이 서버는 **중계기**가 된다(HF Space 용). 비우면 스스로 처리한다(젠 용).
UPSTREAM_ASK = os.environ.get("PHARMA_UPSTREAM_ASK", "").strip()
ACCESS_TOKEN = os.environ.get("PHARMA_ACCESS_TOKEN", "").strip()
ASK_PER_MIN = int(os.environ.get("PHARMA_ASK_PER_MIN", "6"))
_ask_hits: dict[str, list[float]] = {}


def _client_ip(request: Request) -> str:
    fwd = request.headers.get("x-forwarded-for", "")
    if fwd:
        return fwd.split(",")[0].strip()
    return request.client.host if request.client else "?"


def check_access(request: Request):
    """통과하면 None, 막으면 JSONResponse 를 돌려준다."""
    if ACCESS_TOKEN:
        tok = (request.headers.get("x-pharma-access", "")
               or request.query_params.get("k", "")).strip()
        if tok != ACCESS_TOKEN:
            return JSONResponse({"error": "접근 토큰이 필요합니다"}, status_code=401)
    ip = _client_ip(request)
    now = time.time()
    hits = [t for t in _ask_hits.get(ip, []) if now - t < 60]
    if len(hits) >= ASK_PER_MIN:
        _ask_hits[ip] = hits
        return JSONResponse(
            {"error": f"분당 {ASK_PER_MIN}회 제한입니다. 잠시 후 다시 시도하십시오.",
             "retry_after_sec": int(60 - (now - hits[0]))}, status_code=429)
    hits.append(now)
    _ask_hits[ip] = hits
    if len(_ask_hits) > 4096:          # 메모리 무한 증가 방지
        for k in [k for k, v in _ask_hits.items() if not v or now - v[-1] > 300]:
            _ask_hits.pop(k, None)
    return None


class RQ(BaseModel):
    question: str
    max_tokens: int = 900


@app.post("/api/rival")
def api_rival(q: RQ, request: Request):
    """같은 질문을 범용 모델(3사 중 무작위)에 — 비교 데모. 🔴 질문이 외부로 나간다(화면 고지)."""
    blocked = check_access(request)
    if blocked is not None:
        return blocked
    if UPSTREAM_ASK:                        # 릴레이: 상류(젠)가 대신 부른다 — 키는 젠에만 있다
        try:
            body = json.dumps(dict(question=q.question, max_tokens=q.max_tokens)).encode()
            hdrs = {"Content-Type": "application/json",
                    "x-client-ip": _client_ip(request)}
            ut = os.environ.get("PHARMA_UPSTREAM_TOKEN", "").strip()
            if ut:
                hdrs["x-pharma-access"] = ut
            rq2 = urllib.request.Request(UPSTREAM_ASK.replace("/api/ask", "/api/rival"),
                                         data=body, headers=hdrs)
            with urllib.request.urlopen(rq2, timeout=180) as f:
                return JSONResponse(json.load(f))
        except Exception as e:
            return JSONResponse({"error": str(e)[:100]}, status_code=502)
    try:
        import rival
        # 🔴 한도 4,000 — 우리 답과 같은 상한(공정 조건). 900 은 라이벌 답이 중간에 잘려
        #    "일부러 초라하게 보였다"는 시비를 부른다.
        r = rival.answer(q.question, max_tokens=max(100, min(q.max_tokens, 4096)))
        return JSONResponse(r or {"error": "라이벌 응답 없음"})
    except Exception as e:
        return JSONResponse({"error": str(e)[:100]}, status_code=500)


class JQ(BaseModel):
    question: str
    ours: str
    rival_text: str


@app.post("/api/judge")
def api_judge(q: JQ, request: Request):
    """제3자 심판 — 🔴 공개면에서는 **차단**한다(회장님 승인 2026-09-05).
    이유: ①처음 보는 질문에서 승률이 ~50% 라 우리 서비스 페이지에 우리가 지는 화면이 뜬다
          ②같은 답에도 판정이 회차마다 갈린다(측정 불안정)
          ③익명 방문자가 우리 OpenAI 키를 소모한다(과거 OpenRouter 크레딧 소진 전례).
    내부 평가(eval_arena)는 젠에서 직접 호출하므로 영향 없다 — 유일한 외부 정확성 신호로 유지.
    """
    return JSONResponse({"error": "공개면에서는 판정 기능을 제공하지 않습니다"}, status_code=404)
    blocked = check_access(request)
    if blocked is not None:
        return blocked
    if UPSTREAM_ASK:
        try:
            body = json.dumps(dict(question=q.question, ours=q.ours,
                                   rival_text=q.rival_text)).encode()
            hdrs = {"Content-Type": "application/json",
                    "x-client-ip": _client_ip(request)}
            ut = os.environ.get("PHARMA_UPSTREAM_TOKEN", "").strip()
            if ut:
                hdrs["x-pharma-access"] = ut
            rq2 = urllib.request.Request(UPSTREAM_ASK.replace("/api/ask", "/api/judge"),
                                         data=body, headers=hdrs)
            with urllib.request.urlopen(rq2, timeout=240) as f:
                return JSONResponse(json.load(f))
        except Exception as e:
            return JSONResponse({"error": str(e)[:100]}, status_code=502)
    try:
        import rival
        from ask import search as _search
        # 심판용 근거 — 중립 출처(Europe PMC)에서 새로 회수해 대조 기준으로 준다
        ev = ""
        try:
            # 🔴 구조 결함 수리: 심판의 대조 자료가 "자체 검색 6편"뿐이라, 답이 인용한 PMID 가
            #    거기 없으면 실존 논문도 "확인 불가"로 감점됐다(사유문이 실토: "참고 근거가 없어").
            #    ⟹ **두 답이 인용한 PMID 의 초록을 직접 조회해** 대조 자료에 합친다 —
            #       인용 검증이 추측이 아니라 대조가 된다.
            import re as _re2
            cited_ids = list(dict.fromkeys(_re2.findall(r"PMID:(\d{6,9})",
                                                        q.ours + " " + q.rival_text)))[:8]
            ps = []
            for pid in cited_ids:
                try:
                    ps += _search(f"EXT_ID:{pid} AND SRC:MED", n=1)
                except Exception:
                    pass
            have = {str(p["pmid"]) for p in ps}
            ps += [x for x in _search(q.question, n=6) if str(x["pmid"]) not in have]
            # 🔴 초록 900자 절단이 HR·CI 수치를 잘라 심판이 "근거에 없는 수치"로 감점했다 — 1600자로
            ev = (chr(10) * 2).join("[PMID:%s] %s%s%s" % (p["pmid"], p["title"], chr(10),
                                                          p["abstract"][:1600]) for p in ps[:12])
            # 🔴 A 가 기대는 [FDA:약물명] 라벨 원문을 심판이 못 받아 FDA 인용 전부가 "확인 불가"였다
            fda_names = list(dict.fromkeys(_re2.findall(r"\[FDA:([A-Za-z][A-Za-z0-9\- ]{2,30})\]",
                                                        q.ours + " " + q.rival_text)))[:2]
            for nm in fda_names:
                try:
                    import fda as _fda
                    fb = _fda.block(nm.strip())
                    if fb:
                        ev = fb + (chr(10) * 2) + ev
                except Exception:
                    pass
        except Exception:
            pass
        r = rival.judge(q.question, q.ours, q.rival_text, evidence=ev)
        # 🔴 승패 전수 기록 — 이길 때만 세고 질 때 침묵하면 그 승률은 사기다.
        try:
            if r:
                with open(os.path.join(os.path.dirname(os.path.abspath(__file__)),
                                       "judge_ledger.jsonl"), "a", encoding="utf-8") as fp:
                    fp.write(json.dumps(dict(t=time.time(), q=q.question[:200],
                                             winner=r.get("winner"),
                                             reason=(r.get("reason") or "")[:200]),
                                        ensure_ascii=False) + chr(10))
        except Exception:
            pass
        return JSONResponse(r or {"error": "심판 응답 없음"})
    except Exception as e:
        return JSONResponse({"error": str(e)[:100]}, status_code=500)


@app.post("/api/ask")
def api_ask(q: Q, request: Request):
    # 🔴 2026-08-31: HF Space 는 시크릿 없이도 동작해야 한다.
    #    앞서 Space 를 `/api/llm`(토큰 필요)에 붙였더니 **한 번도 동작하지 못했다** —
    #    사람이 시크릿을 넣어줄 때까지 401 이었다. 완성되지 않은 것을 배포로 보고한 셈이다.
    #    ⟹ Space 는 상류의 `/api/ask` 를 **그대로 중계**한다. 토큰이 필요 없고
    #       상류의 분당 제한이 그대로 적용된다(오히려 남용 방지에 낫다).
    if UPSTREAM_ASK:
        try:
            body = json.dumps(dict(question=q.question, mode=q.mode, n=q.n,
                                   web=bool(getattr(q, "web", False)))).encode()
            hdrs = {"Content-Type": "application/json",
                    "x-client-ip": _client_ip(request)}   # 공개 Space: 사용자별 한도를 위해 실IP 전달
            ut = os.environ.get("PHARMA_UPSTREAM_TOKEN", "").strip()
            if ut:
                hdrs["x-pharma-access"] = ut
            rq = urllib.request.Request(UPSTREAM_ASK, data=body, headers=hdrs)
            with urllib.request.urlopen(rq, timeout=600) as f:
                return JSONResponse(json.load(f))
        except urllib.error.HTTPError as e:
            try:
                return JSONResponse(json.load(e), status_code=e.code)
            except Exception:
                return JSONResponse({"error": f"상류 응답 오류 {e.code}"}, status_code=e.code)
        except Exception as e:
            return JSONResponse({"error": f"상류에 닿지 못했습니다: {str(e)[:160]}"},
                                status_code=502)

    blocked = check_access(request)
    if blocked is not None:
        return blocked
    if not q.question.strip():
        return JSONResponse({"error": "질문이 비었습니다"}, status_code=400)
    if q.mode not in THRESHOLDS:
        return JSONResponse({"error": f"mode 는 {sorted(THRESHOLDS)} 중 하나"}, status_code=400)
    t0 = time.time()
    try:
        r = ask(q.question, mode=q.mode, n=max(3, min(q.n, 12)), web=bool(q.web))
    except RuntimeError as e:
        # 설정 누락은 장애가 아니라 **설정 안내**로 돌려준다
        return JSONResponse({"error": str(e)[:300], "setup_required": True}, status_code=503)
    except Exception as e:
        traceback.print_exc()
        return JSONResponse({"error": f"검색 실패: {str(e)[:200]}"}, status_code=500)
    got = {p["pmid"] for p in r["papers"]}
    cited = set(r.get("cited_pmids") or [])
    r["hallucinated_pmids"] = sorted(cited - got)     # 회수 목록 밖 PMID = 환각
    r["elapsed_sec"] = round(time.time() - t0, 1)
    r["threshold"] = THRESHOLDS[q.mode]
    return r


@app.get("/api/health")
def health():
    return dict(ok=True, backend=os.environ.get("PHARMA_BACKEND", "api"),
                proxy=bool(PROXY_TOKEN), rate_per_min=RATE_PER_MIN,
                access_token_required=bool(ACCESS_TOKEN), ask_per_min=ASK_PER_MIN,
                relay=bool(UPSTREAM_ASK),
                model=os.environ.get("PHARMA_LOCAL_MODEL", "-"),
                thresholds=THRESHOLDS)


# ── LLM 프록시 ───────────────────────────────────────────────────────────────
# 🔴 왜 필요한가: HF Space(cpu-basic)는 9B 를 못 돌리고, 젠의 llama-server 는 127.0.0.1 에만
#    묶여 있어 Space 가 닿을 수 없다. 그래서 **Space 는 화면만, 추론은 우리 GPU 에서 우리 모델로.**
#    이것이 없으면 제품이 남의 모델을 쓰게 된다.
#
# 🔴 우리 GPU 를 인터넷에 여는 것이므로 두 가지를 건다:
#    ① 공유 토큰(`PHARMA_PROXY_TOKEN`) — 없으면 프록시 자체가 꺼진다
#    ② 분당 호출 제한 — 공유 GPU 라 남의 작업을 밀어내면 안 된다
PROXY_TOKEN = os.environ.get("PHARMA_PROXY_TOKEN", "").strip()
UPSTREAM = os.environ.get("PHARMA_SERVER", "http://127.0.0.1:7910/v1/chat/completions")
RATE_PER_MIN = int(os.environ.get("PHARMA_RATE_PER_MIN", "20"))
_calls: list[float] = []


@app.post("/api/llm")
async def api_llm(request: Request):
    if not PROXY_TOKEN:
        return JSONResponse({"error": "프록시가 꺼져 있습니다 (PHARMA_PROXY_TOKEN 미설정)"},
                            status_code=503)
    if request.headers.get("x-pharma-token", "") != PROXY_TOKEN:
        return JSONResponse({"error": "인증 실패"}, status_code=401)
    now = time.time()
    _calls[:] = [t for t in _calls if now - t < 60]
    if len(_calls) >= RATE_PER_MIN:
        return JSONResponse({"error": f"분당 {RATE_PER_MIN}회 제한을 넘었습니다"}, status_code=429)
    _calls.append(now)
    body = await request.body()
    rq = urllib.request.Request(UPSTREAM, data=body,
                                headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(rq, timeout=600) as f:
            return JSONResponse(json.load(f))
    except Exception as e:
        return JSONResponse({"error": f"상류 모델 호출 실패: {str(e)[:160]}"}, status_code=502)


# ── 스트리밍 종단점 ──────────────────────────────────────────────────────────
# 검색·게이트는 그대로 두고 **답변 생성만** 흘린다. 게이트는 답이 다 나온 뒤 판정하므로
# 🔴 스트리밍 중에는 아직 **보류 판정 전**이다 — 화면에서 그 사실을 반드시 표시한다.
@app.post("/api/ask/stream")
async def api_ask_stream(q: Q, request: Request):
    blocked = check_access(request)
    if blocked is not None:
        return blocked
    if UPSTREAM_ASK:
        up = UPSTREAM_ASK.replace("/api/ask", "/api/ask/stream")
        def relay():
            body = json.dumps(dict(question=q.question, mode=q.mode, n=q.n,
                                   web=bool(getattr(q, "web", False)))).encode()
            _h = {"Content-Type": "application/json",
                  "x-client-ip": _client_ip(request)}
            _ut = os.environ.get("PHARMA_UPSTREAM_TOKEN", "").strip()
            if _ut:
                _h["x-pharma-access"] = _ut
            rq = urllib.request.Request(up, data=body,
                                        headers=_h)
            try:
                with urllib.request.urlopen(rq, timeout=900) as f:
                    for chunk in f:
                        yield chunk
            except Exception as e:
                yield ("data: " + json.dumps({"type": "error",
                       "error": str(e)[:200]}) + chr(10) + chr(10)).encode()
        return StreamingResponse(relay(), media_type="text/event-stream",
                                 headers={"Cache-Control": "no-store",
                                          "X-Accel-Buffering": "no"})

    def gen():
        def ev(o):
            return "data: " + json.dumps(o, ensure_ascii=False) + chr(10) + chr(10)
        t0 = time.time()
        # 🔴 실사고 2연발의 종지부: 수집을 ask.prepare() **한 함수로 통일**한다.
        #    스트리밍이 자체 수집을 갖던 동안 ①옛 게이트가 좋은 초안을 뒤집었고
        #    ②FDA·UniProt 등 12종 소스가 화면 경로에 빠져 심판전에서 근거 부족으로 졌다.
        try:
            from ask import prepare as A_prepare, LEDGER_ADD as A_LEDGER
            import fulltext as A_FT
            P = A_prepare(q.question, mode=q.mode, n=max(3, min(q.n, 12)),
                          web=bool(getattr(q, "web", False)))
        except Exception as e:
            yield ev({"type": "error", "error": "검색 실패: " + str(e)[:160]}); return
        papers, trials, queries = P["papers"], P["trials"], P["queries"]
        if not papers and not trials:
            yield ev({"type": "done", "answered": False, "reason": "검색 결과 없음",
                      "papers": [], "trials": [], "queries": queries,
                      "elapsed_sec": round(time.time() - t0, 1)}); return
        yield ev({"type": "papers", "papers": papers, "trials": trials, "queries": queries,
                  "qtype": P["qtype"], "used_fda": P["fda"], "used_s2": P["s2"]})
        block = (chr(10) + chr(10)).join(A_FT.block(p) for p in papers)
        if P["extra"]:
            block = block + chr(10) + chr(10) + P["extra"]
        EVB = int(os.environ.get("PHARMA_EVIDENCE_CHARS", "26000"))
        if len(block) > EVB:
            block = block[:EVB] + chr(10) + "(근거가 길어 이후는 잘렸습니다)"
        prompt_t = A_PROMPT + (A_LEDGER if P["qtype"] == "judge" else "")
        prompt = prompt_t.format(q=q.question, papers=block,
                                 trials=(A_CT.block(trials) if trials else "(없음)"))
        buf = []
        try:
            for piece in A_stream_llm(prompt):
                buf.append(piece)
                yield ev({"type": "delta", "text": piece})
        except Exception as e:
            yield ev({"type": "error", "error": "생성 실패: " + str(e)[:160]}); return
        raw = "".join(buf)
        # 🔴 개행이 든 정규식은 파일에 심을 때 자주 깨진다. 문자열 분할로 뽑는다.
        body = raw.split("ANSWER:", 1)[-1]
        ans = body.split("CONFIDENCE:", 1)[0].strip() or raw.strip()
        conf = A_logprob_confidence(block, q.question)
        thr = THRESHOLDS[q.mode]
        # 🔴 실사고(회장님 적발): 스트리밍 경로에 **옛 판정 코드가 따로 살아** 있었다 —
        #    ask.ask() 에 넣은 유형분기·차단권 축소가 여기엔 없어서, 정확한 초안을
        #    다 흘려보낸 뒤 마지막에 옛 게이트가 보류 도장을 찍었다.
        #    같은 판정 규칙을 이식한다: 판단형=게이트 미적용 · 사실형=명백불일치(0.15)만 차단.
        _qt = P["qtype"]
        if _qt == "judge":
            ok = True
        else:
            _blk = float(os.environ.get("PHARMA_FACT_BLOCK", "0.15"))
            ok = conf is not None and conf >= min(thr, _blk)
        # 유령 인용 사후검증 — 실존하면 근거 승격, 아니면 경고 (ask 와 같은 함수)
        try:
            from ask import rescue_citations as A_rescue
            ans, _ghosts, _rescued = A_rescue(ans, papers, trials)
        except Exception:
            _ghosts, _rescued = [], []
        got = {str(p["pmid"]) for p in papers}
        cited = sorted(set(A_re.findall("PMID:([A-Za-z]*[0-9]+)", ans)))
        cited_nct = sorted(set(A_re.findall("NCT:?[ ]*(NCT[0-9]+)", ans)))
        yield ev({"type": "done", "answered": ok, "confidence": conf, "threshold": thr,
                  "qtype": _qt,
                  "answer": ans if ok else None, "withheld_answer": None if ok else ans,
                  "reason": ("판단형 질의 — 게이트 대신 논문별 소견으로 판단하십시오"
                             if _qt == "judge" else
                             (("신뢰도 " + str(round(conf, 3)) + " · 🔴게이트 재보정 중 — 참고만")
                              if ok else ("근거가 질문과 명백히 무관합니다 (신뢰도 "
                                          + str(round(conf, 3) if conf is not None else 0) + ")"))),
                  "cited_pmids": cited, "cited_ncts": cited_nct,
                  "trials": trials,
                  "hallucinated_pmids": _ghosts, "rescued_pmids": _rescued,
                  "papers": papers,
                  "elapsed_sec": round(time.time() - t0, 1)})
    return StreamingResponse(gen(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"})


# ── 혼잡 감지 ────────────────────────────────────────────────────────────────
# 🔴 2026-08-31: 타팀이 같은 GPU 에서 boltz 4프로세스를 돌려 사용률 100% 가 되자
#    "Say OK." 한 마디에 **49초**가 걸렸다. 사용자에게는 **그냥 멈춘 것으로 보인다.**
#    🔴 공유 GPU 이므로 남의 작업은 건드리지 않는다. 대신 **상태를 정직하게 알린다.**
def probe_llm_latency(timeout: int = 40):
    """짧은 요청 하나로 지금 얼마나 밀렸는지 잰다.
    🔴 **타임아웃과 사망을 구분한다.** 25초 안에 답이 없다고 서버가 죽은 것이 아니다 —
       공유 GPU 가 밀리면 "ok" 한 마디도 28~49초가 걸린다(실측).
    🔑 구분은 **예외 종류**로 한다. 타임아웃이 났다면 **연결은 이미 된 것**이고,
       연결 자체가 안 되면 `URLError(ConnectionRefused)` 가 **즉시** 난다.
       (앞서 소켓을 따로 열어 확인하려다 오히려 오판했다 — 단순한 쪽이 옳았다.)
    반환: (초, 상태) · 상태 = ok | timeout | refused
    """
    up = os.environ.get("PHARMA_SERVER", "http://127.0.0.1:7910/v1/chat/completions")
    body = json.dumps(dict(model="local", temperature=0.0, max_tokens=2,
                           chat_template_kwargs={"enable_thinking": False},
                           messages=[{"role": "user", "content": "ok"}])).encode()
    t = time.time()
    try:
        rq = urllib.request.Request(up, data=body,
                                    headers={"Content-Type": "application/json"})
        with urllib.request.urlopen(rq, timeout=timeout) as f:
            f.read(1)
        return round(time.time() - t, 1), "ok"
    except (TimeoutError, socket.timeout):
        return round(time.time() - t, 1), "timeout"
    except urllib.error.URLError as e:
        inner = getattr(e, "reason", None)
        if isinstance(inner, (TimeoutError, socket.timeout)):
            return round(time.time() - t, 1), "timeout"
        return round(time.time() - t, 1), "refused"
    except Exception:
        return round(time.time() - t, 1), "timeout"


@app.get("/api/load")
def api_load():
    """지금 서버가 얼마나 붐비는가. 화면이 검색 전에 물어본다."""
    if UPSTREAM_ASK:
        # 🔴 상류가 밀리면 **부하 조회 자체도 밀린다.** 그것을 `unknown`/`down` 으로 보고하면
        #    가장 붐빌 때 가장 부정확해진다. 조회가 실패한 것 자체가 **혼잡의 신호**다.
        up = UPSTREAM_ASK.replace("/api/ask", "/api/load")
        try:
            with urllib.request.urlopen(urllib.request.Request(up), timeout=60) as f:
                return JSONResponse(json.load(f))
        except urllib.error.URLError as e:
            inner = getattr(e, "reason", None)
            if isinstance(inner, (TimeoutError, socket.timeout)):
                return dict(state="congested", latency_sec=None,
                            note="상류 상태 조회가 60초를 넘겼습니다")
            return dict(state="down", latency_sec=None)
        except Exception:
            return dict(state="congested", latency_sec=None,
                        note="상류 상태를 확인하지 못했습니다")
    ms, st = probe_llm_latency()
    if st == "refused":
        return dict(state="down", latency_sec=None)
    if st == "timeout":
        return dict(state="congested", latency_sec=ms, note="응답이 25초를 넘겼습니다")
    # 한가할 때 이 요청은 1초 미만이다. 실측 혼잡 시 49초.
    state = "idle" if ms < 3 else ("busy" if ms < 15 else "congested")
    return dict(state=state, latency_sec=ms)


@app.get("/", response_class=HTMLResponse)
def index():
    with open(os.path.join(HERE, "index.html"), encoding="utf-8") as f:
        return HTMLResponse(f.read(), headers={"Cache-Control": "no-store"})
        # 🔴 로직 품은 HTML 은 캐시 금지 — 안 그러면 고친 판이 몇 분간 안 뜬다(기록된 사고)


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", "7912")), log_level="info")