프로젝트 3 — 두 서비스를 한 묶음으로 배포하기

import importlib.metadata as md
import shutil, sys
from pathlib import Path
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

plt.rcParams["font.family"] = "Malgun Gothic"        # 한글 글꼴(5단계의 그래프)
plt.rcParams["axes.unicode_minus"] = False
test = pd.read_csv("inquiries.csv")                  # 5단계의 부하 시험에 쓸 문의 40개
summary_bodies = pd.read_json("summary_docs.jsonl", lines=True)["본문"].tolist()

DEPLOY = Path("out_llm_serving")
for name, src, extra in [("summary", "out_summary_api.py", []), ("inquiry", "out_inquiry_api.py", ["out_inquiry_clf.joblib"])]:
    folder = DEPLOY / name
    folder.mkdir(parents=True, exist_ok=True)
    shutil.copy(src, folder / "app.py")                # 프로젝트 1, 2 의 API 를 app.py 로
    for f in extra:
        shutil.copy(f, folder / f)

# 판을 고정한 패키지 목록. 지금 이 PC 에 깔린 판을 적는다
PACKAGES = {"summary": ["fastapi", "uvicorn", "pydantic"],
            "inquiry": ["fastapi", "uvicorn", "pydantic", "joblib", "scikit-learn", "sentence-transformers"]}
for name, pkgs in PACKAGES.items():
    (DEPLOY / name / "requirements.txt").write_text("".join(f"{p}=={md.version(p)}\n" for p in pkgs), encoding="utf-8")

(DEPLOY / "summary" / "Dockerfile").write_text("""FROM python:3.14-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py .
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
""", encoding="utf-8")
(DEPLOY / "inquiry" / "Dockerfile").write_text(f"""FROM python:3.14-slim
WORKDIR /app
# 임베딩은 CPU 에서 돌리므로 CPU 판 PyTorch 를 깐다(GPU 판보다 훨씬 작다)
RUN pip install --no-cache-dir torch=={md.version("torch").split("+")[0]} --index-url https://download.pytorch.org/whl/cpu
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 임베딩 모델을 이미지를 만들 때 받아 둔다. 컨테이너가 뜰 때 인터넷이 없어도 된다
RUN python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('intfloat/multilingual-e5-small')"
COPY app.py out_inquiry_clf.joblib ./
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
""", encoding="utf-8")
for p in sorted(DEPLOY.rglob("*")):
    if p.is_file():
        print(p.as_posix(), f"{p.stat().st_size:,}바이트")
print((DEPLOY / "inquiry" / "requirements.txt").read_text(encoding="utf-8"))

import yaml

compose_text = """services:
  ollama:
    image: ollama/ollama:0.40.12
    volumes:
      - ollama-models:/root/.ollama
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    healthcheck:
      test: ["CMD", "ollama", "list"]
      interval: 10s
      retries: 30
  ollama-pull:
    image: ollama/ollama:0.40.12
    environment:
      OLLAMA_HOST: ollama:11434
    entrypoint: ["ollama", "pull", "qwen2.5:3b"]
    depends_on:
      ollama:
        condition: service_healthy
  summary:
    build: ./summary
    ports: ["8171:8000"]
    environment:
      OLLAMA_URL: http://ollama:11434
      SUMMARY_MODEL: qwen2.5:3b
      SUMMARY_MAX_BUSY: "2"
      SUMMARY_TIMEOUT: "60"
    depends_on:
      ollama-pull:
        condition: service_completed_successfully
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health')"]
      interval: 15s
  inquiry:
    build: ./inquiry
    ports: ["8172:8000"]
    environment:
      OLLAMA_URL: http://ollama:11434
      INQUIRY_MODEL: qwen2.5:3b
      INQUIRY_THRESHOLD: "0.6"
    depends_on:
      ollama-pull:
        condition: service_completed_successfully
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health')"]
      interval: 15s
volumes:
  ollama-models:
"""
(DEPLOY / "compose.yaml").write_text(compose_text, encoding="utf-8")
compose = yaml.safe_load(compose_text)
for name, svc in compose["services"].items():
    print(f"{name:12s} 포트 {svc.get('ports', '-')} · 기다림 {list(svc.get('depends_on', {}))} · GPU {'deploy' in svc}")

import ast, re
from urllib.parse import urlsplit

IMPORT_NAME = {"sklearn": "scikit-learn", "sentence_transformers": "sentence-transformers"}   # import 이름과 설치 이름이 다른 것
problems = []
services = compose["services"]
for name in ["summary", "inquiry"]:
    folder, svc = DEPLOY / name, services[name]
    code = (folder / "app.py").read_text(encoding="utf-8")
    # 1) 앱이 읽는 환경 변수 가운데 compose 가 주는 것과 기본값을 쓰는 것
    used = re.findall(r'os\.environ\.get\("(\w+)"', code)
    given = svc.get("environment", {})
    print(f"{name}: compose 가 주는 값 {[v for v in used if v in given]} · 기본값을 쓰는 값 {[v for v in used if v not in given]}")
    problems += [f"{name}: 앱이 읽지 않는 {k}" for k in given if k not in used]
    # 2) 주소의 호스트가 compose 의 서비스 이름인가
    problems += [f"{name}.{k}: {urlsplit(v).hostname} 서비스가 없다" for k, v in given.items()
                 if str(v).startswith("http") and urlsplit(v).hostname not in services]
    # 3) Dockerfile 의 포트와 compose 의 컨테이너 쪽 포트
    if '"--port", "8000"' not in (folder / "Dockerfile").read_text(encoding="utf-8") or not svc["ports"][0].endswith(":8000"):
        problems.append(f"{name}: 포트가 맞지 않는다")
    # 4) import 한 바깥 패키지가 requirements.txt 에 있나
    tree = ast.parse(code)
    mods = {a.name.split(".")[0] for n in ast.walk(tree) if isinstance(n, ast.Import) for a in n.names}
    mods |= {n.module.split(".")[0] for n in ast.walk(tree) if isinstance(n, ast.ImportFrom) and n.module}
    third = {IMPORT_NAME.get(m, m) for m in mods if m not in sys.stdlib_module_names}
    reqs = {line.split("==")[0] for line in (folder / "requirements.txt").read_text(encoding="utf-8").split()}
    problems += [f"{name}: requirements.txt 에 {p} 가 없다" for p in third - reqs]
gpu = [n for n, s in services.items() if "deploy" in s]
print("GPU 를 받는 서비스:", gpu, "· 모델 볼륨:", services["ollama"]["volumes"])
print("문제:", problems or "없음")

import json, os, subprocess, time, urllib.error, urllib.request

servers = {}
for name, port in [("summary", 8171), ("inquiry", 8172)]:
    env = {k: str(v) for k, v in services[name]["environment"].items()}
    env["OLLAMA_URL"] = "http://127.0.0.1:11434"                   # compose 라면 http://ollama:11434
    servers[name] = subprocess.Popen([sys.executable, "-m", "uvicorn", "app:app", "--port", str(port), "--log-level", "warning"],
                                     cwd=DEPLOY / name, env=dict(os.environ, **env))
for name, port in [("summary", 8171), ("inquiry", 8172)]:
    for _ in range(240):                                          # 데우기까지 끝나야 /health 가 답한다
        if servers[name].poll() is not None:                      # 켜다가 멈췄으면 더 기다리지 않는다
            print(name, "켜지 못함")
            break
        try:
            print(name, urllib.request.urlopen(f"http://127.0.0.1:{port}/health", timeout=2).read().decode())
            break
        except OSError:
            time.sleep(0.5)
print("문서 화면:", urllib.request.urlopen("http://127.0.0.1:8172/docs").status)

from concurrent.futures import ThreadPoolExecutor


def call(url, body, tries=4):
    """다른 팀이 쓸 부르는 쪽 코드. 503 이면 Retry-After 만큼 기다렸다 다시 보낸다. (상태, 본문, 보낸 횟수)"""
    data = json.dumps(body, ensure_ascii=False).encode()
    for i in range(tries):
        req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"})
        start = time.perf_counter()
        try:
            with urllib.request.urlopen(req, timeout=120) as r:
                return r.status, json.loads(r.read()), i + 1, time.perf_counter() - start
        except urllib.error.HTTPError as e:
            if e.code != 503:
                return e.code, json.loads(e.read()), i + 1, time.perf_counter() - start
            time.sleep(float(e.headers.get("Retry-After", 5)))
    return 503, None, tries, None


questions = test["문의"].tolist()
load = []
for workers in [1, 4, 8]:
    t = time.perf_counter()
    with ThreadPoolExecutor(max_workers=workers) as pool:          # 다른 팀 여럿이 한꺼번에 부르는 흉내
        res = list(pool.map(lambda q: call("http://127.0.0.1:8172/classify", {"문의": q}), questions))
    total = time.perf_counter() - t
    lat = [r[3] for r in res]
    ok = np.mean([r[1]["분류"] == y for r, y in zip(res, test["분류"])])
    load.append({"동시": workers, "초당 처리": round(40 / total, 2), "p50": round(float(np.percentile(lat, 50)), 2),
                 "p95": round(float(np.percentile(lat, 95)), 2), "전체 초": round(total, 1), "정확도": round(float(ok), 3),
                 "LLM": sum(r[1]["판단"] == "LLM" for r in res)})
print(pd.DataFrame(load).to_string(index=False))

# 요약 API: 넷이 한꺼번에 보내도, 부르는 쪽이 Retry-After 를 지키면 모두 답을 받는다
with ThreadPoolExecutor(max_workers=4) as pool:
    res = list(pool.map(lambda d: call("http://127.0.0.1:8171/summarize", {"본문": d, "문장수": 1}), summary_bodies[:4]))
print("요약 넷: 상태", [r[0] for r in res], "· 보낸 횟수", [r[2] for r in res])

fig, ax = plt.subplots(1, 2, figsize=(10, 4))
d = pd.DataFrame(load)
ax[0].bar(d["동시"].astype(str), d["초당 처리"])
ax[0].set_title("분류 API 의 초당 처리 수")
ax[0].set_xlabel("동시 요청 수")
ax[1].plot(d["동시"], d["p50"], marker="o", label="p50")
ax[1].plot(d["동시"], d["p95"], marker="o", label="p95")
ax[1].set_title("요청 하나의 응답 시간(초)")
ax[1].set_xlabel("동시 요청 수")
ax[1].legend()
plt.tight_layout()
plt.show()

for name, proc in servers.items():                               # 띄운 서버를 끈다
    proc.terminate()
    proc.wait(timeout=10)
print("서버를 껐습니다:", list(servers))