프로젝트 1 — 온프레미스 LLM 서빙 서비스

import json, os, secrets, shutil, subprocess, sys, time, urllib.request
from importlib.metadata import version
from pathlib import Path

SERVICE = Path("out_service")
shutil.rmtree(SERVICE, ignore_errors=True)
(SERVICE / "api").mkdir(parents=True)
(SERVICE / "ui").mkdir(parents=True)
shutil.copy("company_rules.txt", SERVICE / "api")

(SERVICE / "api" / "main.py").write_text('''"""사내 규정 질의 API(13차시 프로젝트를 설정 가능하게 바꾼 것)"""
import hashlib, hmac, json, os, re, time
import urllib.request
from collections import defaultdict

import numpy as np
from fastapi import Depends, FastAPI, HTTPException, Security
from fastapi.security import APIKeyHeader
from pydantic import BaseModel, Field
from sentence_transformers import SentenceTransformer

OLLAMA = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
MODEL = os.environ.get("CHAT_MODEL", "qwen2.5:3b")
RULES = os.environ.get("RULES_FILE", "company_rules.txt")
digest = lambda k: hashlib.sha256(k.encode()).hexdigest()
KEYS = {digest(v["키"]): {"이름": n, "역할": v["역할"]} for n, v in json.loads(os.environ["RULES_API_KEYS"]).items()}

articles = [l.strip() for l in open(RULES, encoding="utf-8") if re.match(r"^제\\d+조\\(", l)]
titles = [re.match(r"^제\\d+조\\((.+?)\\)", a).group(1) for a in articles]
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
A = emb.encode(["passage: " + a for a in articles], normalize_embeddings=True)
SYSTEM = "당신은 솔내기술(가상 회사)의 규정 도우미입니다. [조항]에 있는 내용만으로 한두 문장으로 답하세요. 없으면 '규정에서 찾을 수 없습니다'라고 답하세요."
recent = defaultdict(list)
app = FastAPI(title="사내 규정 질의 API")
header = APIKeyHeader(name="X-API-Key", auto_error=False)


def who(key: str | None = Security(header)):
    user = next((u for h, u in KEYS.items() if key and hmac.compare_digest(h, digest(key))), None)
    if user is None:
        raise HTTPException(status_code=401, detail="API 키가 없거나 맞지 않습니다")
    now = time.time()
    recent[user["이름"]] = [t for t in recent[user["이름"]] if now - t < 60]
    if len(recent[user["이름"]]) >= 10:
        raise HTTPException(status_code=429, detail="1분에 10번까지 부를 수 있습니다")
    recent[user["이름"]].append(now)
    return user


def llm(messages):
    body = {"model": MODEL, "messages": messages, "stream": False, "options": {"temperature": 0, "seed": 42}}
    req = urllib.request.Request(OLLAMA + "/api/chat", data=json.dumps(body).encode(), headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.loads(r.read())["message"]["content"].strip()


class Ask(BaseModel):
    질문: str = Field(min_length=2, max_length=200)


@app.get("/health")
def health():
    return {"상태": "ok", "조항": len(articles), "모델": MODEL}


@app.post("/ask")
def ask(body: Ask, user=Depends(who)):
    start = time.perf_counter()
    q = emb.encode("query: " + body.질문, normalize_embeddings=True)
    top = np.argsort(-(A @ q))[:3]
    text = llm([{"role": "system", "content": SYSTEM},
                {"role": "user", "content": "[조항]\\n" + "\\n".join(articles[i] for i in top) + "\\n\\n[질문]\\n" + body.질문}])
    return {"답": text, "근거조항": [titles[i] for i in top], "걸린ms": round((time.perf_counter() - start) * 1000)}
''', encoding="utf-8")

pkgs = ["fastapi", "uvicorn", "numpy", "sentence-transformers"]
(SERVICE / "api" / "requirements.txt").write_text("\n".join(f"{p}=={version(p)}" for p in pkgs) + "\n", encoding="utf-8")
(SERVICE / "api" / "Dockerfile").write_text("""FROM python:3.14-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8014
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8014"]
""", encoding="utf-8")
print(sorted(p.name for p in (SERVICE / "api").iterdir()))
print((SERVICE / "api" / "requirements.txt").read_text(encoding="utf-8"))

(SERVICE / "ui" / "app.py").write_text('''"""사내 규정 채팅 화면"""
import json
import os
import urllib.error
import urllib.request

import streamlit as st

API = os.environ.get("API_URL", "http://127.0.0.1:8014")
KEY = os.environ.get("UI_API_KEY", "")                 # 화면 서버만 아는 키. 브라우저로는 보내지 않는다
MESSAGES = {401: "화면의 API 키가 맞지 않습니다. 관리자에게 알려 주세요.",
            429: "질문이 너무 잦습니다. 1분 뒤에 다시 물어 주세요.",
            422: "질문은 2글자에서 200글자 사이로 써 주세요."}


def ask(question):
    req = urllib.request.Request(API + "/ask", data=json.dumps({"질문": question}, ensure_ascii=False).encode(),
                                 headers={"Content-Type": "application/json", "X-API-Key": KEY})
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.loads(r.read())


st.title("솔내기술 규정 도우미")
st.caption("사내 규정집에서 찾아 답합니다. 회사와 규정은 연습용으로 지어낸 것입니다.")
if "messages" not in st.session_state:
    st.session_state.messages = []
for m in st.session_state.messages:
    with st.chat_message(m["role"]):
        st.write(m["content"])
        if m.get("근거"):
            st.caption("근거 조항: " + ", ".join(m["근거"]))

if prompt := st.chat_input("규정을 물어보세요"):
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.write(prompt)
    with st.chat_message("assistant"):
        try:
            with st.spinner("규정을 찾는 중"):
                result = ask(prompt)
        except urllib.error.HTTPError as e:
            st.error(MESSAGES.get(e.code, f"요청이 거절됐습니다({e.code})."))
            st.stop()
        except OSError:
            st.error("규정 API 에 연결할 수 없습니다. 잠시 뒤에 다시 해 주세요.")
            st.stop()
        st.write(result["답"])
        st.caption("근거 조항: " + ", ".join(result["근거조항"]))
    st.session_state.messages.append({"role": "assistant", "content": result["답"], "근거": result["근거조항"]})
''', encoding="utf-8")
(SERVICE / "ui" / "requirements.txt").write_text(f"streamlit=={version('streamlit')}\n", encoding="utf-8")
(SERVICE / "ui" / "Dockerfile").write_text("""FROM python:3.14-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8501
CMD ["streamlit", "run", "app.py", "--server.port", "8501", "--server.address", "0.0.0.0", "--server.headless", "true", "--browser.gatherUsageStats", "false"]
""", encoding="utf-8")
print(sorted(p.name for p in (SERVICE / "ui").iterdir()))

(SERVICE / "compose.yaml").write_text("""services:
  ollama:
    image: ollama/ollama
    volumes: ["ollama:/root/.ollama"]
  api:
    build: ./api
    environment:
      OLLAMA_URL: http://ollama:11434
      RULES_API_KEYS: ${RULES_API_KEYS}
    depends_on: [ollama]
  ui:
    build: ./ui
    ports: ["8501:8501"]
    environment:
      API_URL: http://api:8014
      UI_API_KEY: ${UI_API_KEY}
    depends_on: [api]
volumes:
  ollama: {}
""", encoding="utf-8")
# .env 는 비밀을 담으므로 저장소에 올리지 않는다. 꼴만 보이는 견본을 둔다
(SERVICE / ".env.example").write_text('RULES_API_KEYS={"화면": {"키": "발급받은_인증키", "역할": "직원"}}\nUI_API_KEY=발급받은_인증키\n', encoding="utf-8")
import yaml
compose = yaml.safe_load((SERVICE / "compose.yaml").read_text(encoding="utf-8"))
print({n: {"ports": s.get("ports"), "env": sorted(s.get("environment", {}))} for n, s in compose["services"].items()})
print("compose 에 비밀 원문이 있나:", "키\":" in (SERVICE / "compose.yaml").read_text(encoding="utf-8"))

from streamlit.testing.v1 import AppTest

# Ollama 모델을 내렸다 다시 올린다(10차시)
urllib.request.urlopen(urllib.request.Request("http://localhost:11434/api/generate", headers={"Content-Type": "application/json"},
                                              data=json.dumps({"model": "qwen2.5:3b", "keep_alive": 0}).encode())).read()
ui_key = secrets.token_urlsafe(24)                                       # 시험용 키를 그 자리에서 만든다
env = dict(os.environ, RULES_API_KEYS=json.dumps({"화면": {"키": ui_key, "역할": "직원"}}, ensure_ascii=False))
api = subprocess.Popen([sys.executable, "-m", "uvicorn", "main:app", "--port", "8014", "--log-level", "warning"],
                       cwd=SERVICE / "api", env=env, stderr=subprocess.DEVNULL)
t = time.perf_counter()
for _ in range(120):
    try:
        print(urllib.request.urlopen("http://127.0.0.1:8014/health").read().decode(), f"· {time.perf_counter() - t:.1f}초 만에 준비")
        break
    except OSError:
        time.sleep(0.5)

os.environ.update(API_URL="http://127.0.0.1:8014", UI_API_KEY=ui_key)
at = AppTest.from_file(str((SERVICE / "ui" / "app.py").resolve()), default_timeout=180).run()
qa = [json.loads(line) for line in open("company_rules_qa.jsonl", encoding="utf-8")][8:11]
for item in qa:
    at.chat_input[0].set_value(item["질문"]).run()
    reply = at.chat_message[-1]
    ok = any(k in reply.markdown[0].value.replace(" ", "") for k in item["정답열쇠"].replace(" ", "").split("|"))
    print("O" if ok else "X", item["질문"], "→", reply.markdown[0].value, "|", reply.caption[0].value)

os.environ["UI_API_KEY"] = "wrong-key-0000"
bad = AppTest.from_file(str((SERVICE / "ui" / "app.py").resolve()), default_timeout=60).run()
bad.chat_input[0].set_value("연차는 며칠이에요?").run()
print("틀린 키:", [e.value for e in bad.error])
api.terminate()
api.wait(timeout=10)
down = AppTest.from_file(str((SERVICE / "ui" / "app.py").resolve()), default_timeout=60).run()
down.chat_input[0].set_value("연차는 며칠이에요?").run()
print("API 꺼짐:", [e.value for e in down.error])
print(sorted(p.relative_to(SERVICE).as_posix() for p in SERVICE.rglob("*") if p.is_file()))