""" 제약 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")