프로젝트 1 — 요약 API 서비스

import json, re, time, urllib.error, urllib.request
from pathlib import Path
import numpy as np

docs = [json.loads(line) for line in Path("summary_docs.jsonl").read_text(encoding="utf-8").splitlines()]
print(len(docs), "건 ·", {k: sum(d["종류"] == k for d in docs) for k in ["회의록", "공지", "리뷰 묶음", "장애 보고"]})
print("본문 길이:", min(len(d["본문"]) for d in docs), "~", max(len(d["본문"]) for d in docs), "자 · 1번 핵심:", docs[0]["핵심"])

OLLAMA = "http://127.0.0.1:11434"
MODEL = "qwen2.5:3b"
BROKEN = re.compile(r"(.)\1{9,}")            # 같은 글자가 10번 넘게 이어지면 깨진 답(16차시)
retried = {"수": 0}


def post(path, body, timeout=120):
    req = urllib.request.Request(OLLAMA + path, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read())


def ask(messages, num_predict=300):
    """Ollama 에 한 번 묻는다. 500 오류나 깨진 답이면 모델을 내렸다 다시 묻는다(16차시)"""
    body = {"model": MODEL, "messages": messages, "stream": False,
            "options": {"temperature": 0, "seed": 42, "num_predict": num_predict}}
    for _ in range(3):
        try:
            out = post("/api/chat", body)["message"]["content"].strip()
            if not BROKEN.search(out):
                return out
        except urllib.error.HTTPError:
            pass
        retried["수"] += 1
        post("/api/generate", {"model": MODEL, "keep_alive": 0})     # 모델을 내린다
        time.sleep(2)
    return out


def split_sentences(text):
    text = re.sub(r"\s+", " ", text).strip()
    return [s for s in re.split(r"(?<=[.!?])\s+", text) if s]


def score(doc, summary):
    """핵심 사실 셋 가운데 요약에 든 수"""
    return sum(any(w in summary for w in key.split("|")) for key in doc["핵심"])


ask([{"role": "user", "content": "안녕하세요"}], num_predict=5)      # 몸풀기: 모델을 GPU 에 올린다
base = []
for d in docs:                                                       # 노트북에서 하던 그대로 묻는다
    t = time.perf_counter()
    s = ask([{"role": "user", "content": "다음 글을 3문장으로 요약해 주세요.\n\n" + d["본문"]}])
    base.append({"번호": d["번호"], "요약": s, "핵심": score(d, s), "문장": len(split_sentences(s)), "초": time.perf_counter() - t})
print(f"기준선: 핵심 {sum(b['핵심'] for b in base)}/36 · 3문장을 지킴 {sum(b['문장'] == 3 for b in base)}/12 · "
      f"평균 {np.mean([b['초'] for b in base]):.2f}초 · 문장 수 {sorted(b['문장'] for b in base)} · 다시 물음 {retried['수']}")
print("1번 요약:", base[0]["요약"].replace("\n", " ")[:160])

SUMMARY_API = r'''
"""솔빛생활 사내 문서 요약 API (17차시 프로젝트 1)"""
import json, os, re, threading, time, urllib.error, urllib.request
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

# 설정은 환경 변수에서 읽는다. 없으면 오른쪽의 기본값
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
MODEL = os.environ.get("SUMMARY_MODEL", "qwen2.5:3b")
MAX_BUSY = int(os.environ.get("SUMMARY_MAX_BUSY", "2"))       # 한꺼번에 처리할 요청 수
TIMEOUT = float(os.environ.get("SUMMARY_TIMEOUT", "60"))      # 모델을 기다릴 최대 초
VERSION = "summary-1.0"
SYSTEM = ("당신은 사내 문서를 요약하는 도우미입니다. 글에 있는 사실만 써서 정확히 {n}문장으로 요약하세요. "
          "날짜와 숫자는 글에 있는 그대로 쓰세요. 머리말이나 목록 없이 문장만 쓰세요.")
BROKEN = re.compile(r"(.)\1{9,}")

state = {"준비": False, "데우기초": None, "데우기실패": False, "다시물음": 0, "캐시적중": 0}
slots = threading.BoundedSemaphore(MAX_BUSY)
cache = {}


def post(path, body, timeout):
    req = urllib.request.Request(OLLAMA_URL + path, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read())


def generate(text, n, timeout):
    """모델에 요약을 시킨다. 500 오류나 깨진 답이면 모델을 내렸다 두 번까지 다시 묻는다"""
    body = {"model": MODEL, "stream": False, "keep_alive": "30m",
            "messages": [{"role": "system", "content": SYSTEM.format(n=n)}, {"role": "user", "content": text}],
            "options": {"temperature": 0, "seed": 42, "num_predict": 100 * n}}
    for _ in range(3):
        try:
            out = post("/api/chat", body, timeout)["message"]["content"].strip()
            if not BROKEN.search(out):
                return out
        except urllib.error.HTTPError:
            pass
        state["다시물음"] += 1
        post("/api/generate", {"model": MODEL, "keep_alive": 0}, 30)
        time.sleep(2)
    raise RuntimeError("모델이 세 번 모두 답하지 못했습니다")


def split_sentences(text):
    text = re.sub(r"\s+", " ", text).strip()
    return [s for s in re.split(r"(?<=[.!?])\s+", text) if s]


@asynccontextmanager
async def lifespan(app):
    t = time.perf_counter()
    try:                                                    # 데우기: 모델을 GPU 에 올린다
        generate("서버를 켭니다. 모델을 올려 두려고 보내는 글입니다.", 1, 300)
    except (OSError, RuntimeError):                         # 데우기에 실패해도 서버는 뜬다. 첫 요청이 대신 늦을 뿐이다
        state["데우기실패"] = True
    state["데우기초"] = round(time.perf_counter() - t, 2)
    state["준비"] = True
    yield
    state["준비"] = False


app = FastAPI(title="사내 문서 요약 API", version=VERSION, lifespan=lifespan)


class SummaryIn(BaseModel):
    본문: str = Field(min_length=50, max_length=3000, description="요약할 글")
    문장수: int = Field(default=3, ge=1, le=5, description="요약의 문장 수")


class SummaryOut(BaseModel):
    요약: str
    문장수: int
    모델문장수: int
    모델: str
    판: str
    캐시: bool
    걸린초: float


@app.get("/health")
def health():
    return {**state, "모델": MODEL, "판": VERSION, "동시처리": MAX_BUSY, "시간제한": TIMEOUT}


@app.post("/summarize", response_model=SummaryOut)
def summarize(req: SummaryIn):
    t = time.perf_counter()
    key = (" ".join(req.본문.split()), req.문장수)          # 빈칸만 다른 글은 같은 글로 본다
    if key in cache:
        state["캐시적중"] += 1
        return SummaryOut(**cache[key], 캐시=True, 걸린초=round(time.perf_counter() - t, 3))
    if not slots.acquire(blocking=False):                   # 화구가 모두 차 있으면 바로 돌려보낸다
        raise HTTPException(503, "요청이 많아 지금은 받을 수 없습니다. 잠시 뒤 다시 보내 주세요.",
                            headers={"Retry-After": "5"})
    try:
        raw = generate(req.본문, req.문장수, TIMEOUT)
    except TimeoutError:
        raise HTTPException(504, f"모델이 {TIMEOUT:g}초 안에 답하지 못했습니다.")
    except (urllib.error.URLError, RuntimeError) as e:
        raise HTTPException(502, f"모델 서버 오류: {e}")
    finally:
        slots.release()
    sentences = split_sentences(raw)
    result = {"요약": " ".join(sentences[:req.문장수]), "문장수": min(len(sentences), req.문장수),
              "모델문장수": len(sentences), "모델": MODEL, "판": VERSION}
    if len(cache) >= 1000:                                  # 캐시가 차면 가장 오래된 것부터 지운다
        cache.pop(next(iter(cache)))
    cache[key] = result
    return SummaryOut(**result, 캐시=False, 걸린초=round(time.perf_counter() - t, 3))
'''
Path("out_summary_api.py").write_text(SUMMARY_API.lstrip(), encoding="utf-8")
print("out_summary_api.py ·", len(SUMMARY_API.splitlines()), "줄")

import sys
sys.path.insert(0, ".")                    # 작업 폴더의 out_summary_api.py 를 불러오려고
from fastapi.testclient import TestClient
import out_summary_api as api

with TestClient(api.app) as client:        # with 로 열면 lifespan 의 데우기가 먼저 돈다
    print("health:", client.get("/health").json())
    doc = docs[0]["본문"]
    first = client.post("/summarize", json={"본문": doc, "문장수": 2}).json()
    again = client.post("/summarize", json={"본문": doc.replace(" ", "  "), "문장수": 2}).json()   # 빈칸만 다른 글
    print("첫 요청:", {k: first[k] for k in ["문장수", "모델문장수", "캐시", "걸린초"]})
    print("같은 글:", {k: again[k] for k in ["문장수", "모델문장수", "캐시", "걸린초"]}, "· 요약 같음:", first["요약"] == again["요약"])
    print("짧은 글:", client.post("/summarize", json={"본문": "너무 짧아요"}).status_code,
          "· 문장수 9:", client.post("/summarize", json={"본문": doc, "문장수": 9}).status_code,
          "· 본문 없음:", client.post("/summarize", json={"문장수": 2}).status_code)
    print("요약:", first["요약"])

with TestClient(api.app) as client:
    rows = []
    for d in docs:
        r = client.post("/summarize", json={"본문": d["본문"], "문장수": 3}).json()
        rows.append({"번호": d["번호"], "요약": r["요약"], "핵심": score(d, r["요약"]), "문장": r["문장수"],
                     "모델문장": r["모델문장수"], "초": r["걸린초"]})
    health = client.get("/health").json()

print(f"기준선: 핵심 {sum(b['핵심'] for b in base)}/36 · 3문장 {sum(b['문장'] == 3 for b in base)}/12 · 평균 {np.mean([b['초'] for b in base]):.2f}초")
print(f"API   : 핵심 {sum(r['핵심'] for r in rows)}/36 · 3문장 {sum(r['문장'] == 3 for r in rows)}/12 · 평균 {np.mean([r['초'] for r in rows]):.2f}초")
print("모델이 쓴 문장 수:", [r["모델문장"] for r in rows], "· 다시 물음:", health["다시물음"])
for r in rows:
    if r["핵심"] < 3:                                     # 핵심을 놓친 요약을 읽어 본다
        print(f"  {r['번호']}번 핵심 {r['핵심']}/3:", r["요약"][:170])

import os, subprocess
from concurrent.futures import ThreadPoolExecutor


def start(port, **env):
    """out_summary_api 를 uvicorn 으로 띄우고 /health 가 답할 때까지 기다린다(데우기 포함)"""
    proc = subprocess.Popen([sys.executable, "-m", "uvicorn", "out_summary_api:app", "--port", str(port), "--log-level", "warning"],
                            env=dict(os.environ, **env))
    for _ in range(240):
        if proc.poll() is not None:                       # 켜다가 멈췄으면 더 기다리지 않는다
            raise RuntimeError("서버를 켜지 못했습니다")
        try:
            urllib.request.urlopen(f"http://127.0.0.1:{port}/health", timeout=2)
            return proc
        except OSError:
            time.sleep(0.5)


def call(port, body):
    """POST /summarize 를 보내고 (상태 코드, 본문, Retry-After) 를 돌려준다"""
    req = urllib.request.Request(f"http://127.0.0.1:{port}/summarize", data=json.dumps(body, ensure_ascii=False).encode(),
                                 headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=120) as r:
            return r.status, json.loads(r.read()), None
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read()), e.headers.get("Retry-After")


server = start(8171, SUMMARY_MAX_BUSY="2")
bodies = [{"본문": d["본문"], "문장수": 2} for d in docs[4:8]]          # 서로 다른 글 넷(캐시에 없는 글)
with ThreadPoolExecutor(max_workers=4) as pool:                       # 넷을 한꺼번에 보낸다
    results = list(pool.map(lambda b: call(8171, b), bodies))
print("넷을 한꺼번에:", sorted(r[0] for r in results))
busy = [r for r in results if r[0] == 503][0]
print("503 의 본문과 Retry-After:", busy[1]["detail"], "·", busy[2])
time.sleep(int(busy[2]))
print("5초 뒤 다시:", call(8171, bodies[results.index(busy)])[0])
server.terminate(); server.wait(timeout=10)

slow = start(8172, SUMMARY_TIMEOUT="1")                               # 시간 제한을 1초로 줄인 서버
status, body, _ = call(8172, {"본문": docs[10]["본문"], "문장수": 3})
print("시간 제한 1초:", status, body["detail"])
slow.terminate(); slow.wait(timeout=10)