프로젝트 2 — 문의 분류 API 서비스

import json, sys, time, urllib.request
import joblib
import numpy as np
import pandas as pd
from sentence_transformers import SentenceTransformer
from sklearn.linear_model import LogisticRegression

train = pd.read_csv("inquiry_train.csv")
test = pd.read_csv("inquiries.csv")
print("학습", len(train), "개 · 시험", len(test), "개 · 학습 예:", train["문의"].head(3).tolist())

# 임베딩은 CPU 에 둔다. GPU 는 Ollama 가 쓴다(7차시)
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
encode = lambda texts: emb.encode(["query: " + t for t in texts], normalize_embeddings=True)
clf = LogisticRegression(max_iter=1000, C=10).fit(encode(train["문의"].tolist()), train["분류"])

t = time.perf_counter()
proba = clf.predict_proba(encode(test["문의"].tolist()))          # 문의마다 다섯 분류의 확률
fast_s = time.perf_counter() - t
pred = clf.classes_[proba.argmax(axis=1)]
conf = proba.max(axis=1)                                          # 확신도
print(f"분류기 정확도 {(pred == test['분류']).mean():.3f} · 40개에 {fast_s * 1000:.0f}ms · "
      f"확신도 p10/p50/p90 {np.round(np.percentile(conf, [10, 50, 90]), 3).tolist()}")

# 분류기와 판을 한 파일로 저장한다. API 가 켜질 때 읽는다
joblib.dump({"분류기": clf, "임베딩": "intfloat/multilingual-e5-small", "판": "inquiry-lr-1"}, "out_inquiry_clf.joblib")

LABELS = ["배송", "교환반품", "결제", "회원", "기타"]
SCHEMA = {"type": "object", "properties": {"분류": {"type": "string", "enum": LABELS}}, "required": ["분류"]}
SYSTEM = ("쇼핑몰 고객 문의를 다섯 분류 가운데 하나로 나누세요.\n"
          "- 배송: 배송 상태, 주소, 택배\n- 교환반품: 교환, 반품, 불량, 잘못 온 상품\n"
          "- 결제: 결제, 할부, 영수증, 환불 금액, 쿠폰 할인\n- 회원: 로그인, 계정, 등급, 적립금, 회원 쿠폰\n"
          "- 기타: 재입고, 포장, 매장, 리뷰, 고객센터 같은 그 밖의 문의\nJSON 으로 답하세요.")
retried = {"수": 0}


def post(path, body, timeout=60):
    req = urllib.request.Request("http://127.0.0.1:11434" + 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_llm(text):
    """format 에 준 JSON 스키마의 enum 때문에 다섯 분류 가운데 하나만 나온다. 오류면 모델을 내렸다 다시 묻는다"""
    body = {"model": "qwen2.5:3b", "stream": False, "format": SCHEMA,
            "messages": [{"role": "system", "content": SYSTEM}, {"role": "user", "content": text}],
            "options": {"temperature": 0, "seed": 42, "num_predict": 30}}
    for _ in range(3):
        try:
            return json.loads(post("/api/chat", body)["message"]["content"])["분류"]
        except (OSError, ValueError, KeyError):
            retried["수"] += 1
            post("/api/generate", {"model": "qwen2.5:3b", "keep_alive": 0})
            time.sleep(2)
    return None


ask_llm("몸풀기 문의입니다")
t = time.perf_counter()
llm = np.array([ask_llm(q) for q in test["문의"]])
llm_s = time.perf_counter() - t
print(f"LLM 정확도 {(llm == test['분류']).mean():.3f} · 40개에 {llm_s:.1f}초(하나에 {llm_s / 40:.2f}초) · 다시 물음 {retried['수']}")
for q, y, a, b in zip(test["문의"], test["분류"], pred, llm):
    if b != y:
        print(f"  LLM 이 틀림: {q} 정답 {y} · LLM {b} · 분류기 {a}")

rows = []
for th in [0.4, 0.5, 0.6, 0.7, 0.8, 0.9]:
    to_llm = conf < th                                       # 확신도가 기준값보다 낮으면 LLM 에 넘긴다
    final = np.where(to_llm, llm, pred)
    rows.append({"기준값": th, "LLM 에 넘김": int(to_llm.sum()), "정확도": round(float((final == test["분류"]).mean()), 3),
                 "분류기가 끝낸 것의 정확도": round(float((pred[~to_llm] == test["분류"].values[~to_llm]).mean()), 3) if (~to_llm).any() else None,
                 "40개 예상 초": round(fast_s + to_llm.sum() * llm_s / 40, 1)})
print(pd.DataFrame(rows).to_string(index=False))
print(f"(견주기) 분류기만 {(pred == test['분류']).mean():.3f}, {fast_s:.2f}초 · LLM 만 {(llm == test['분류']).mean():.3f}, {llm_s:.1f}초")

INQUIRY_API = r'''
"""솔빛생활 고객 문의 분류 API (17차시 프로젝트 2)"""
import json, os, time, urllib.request
from contextlib import asynccontextmanager
from typing import Annotated

import joblib
from fastapi import FastAPI
from pydantic import BaseModel, Field
from sentence_transformers import SentenceTransformer

OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
MODEL = os.environ.get("INQUIRY_MODEL", "qwen2.5:3b")
THRESHOLD = float(os.environ.get("INQUIRY_THRESHOLD", "0.6"))     # 확신도가 이보다 낮으면 LLM 에 넘긴다
CLF_PATH = os.environ.get("INQUIRY_CLF", "out_inquiry_clf.joblib")
LABELS = ["배송", "교환반품", "결제", "회원", "기타"]
SCHEMA = {"type": "object", "properties": {"분류": {"type": "string", "enum": LABELS}}, "required": ["분류"]}
SYSTEM = ("쇼핑몰 고객 문의를 다섯 분류 가운데 하나로 나누세요.\n"
          "- 배송: 배송 상태, 주소, 택배\n- 교환반품: 교환, 반품, 불량, 잘못 온 상품\n"
          "- 결제: 결제, 할부, 영수증, 환불 금액, 쿠폰 할인\n- 회원: 로그인, 계정, 등급, 적립금, 회원 쿠폰\n"
          "- 기타: 재입고, 포장, 매장, 리뷰, 고객센터 같은 그 밖의 문의\nJSON 으로 답하세요.")
state = {"준비": False, "데우기실패": False, "LLM호출": 0, "LLM실패": 0}


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 ask_llm(text, timeout=30):
    body = {"model": MODEL, "stream": False, "keep_alive": "30m", "format": SCHEMA,
            "messages": [{"role": "system", "content": SYSTEM}, {"role": "user", "content": text}],
            "options": {"temperature": 0, "seed": 42, "num_predict": 30}}
    return json.loads(post("/api/chat", body, timeout)["message"]["content"])["분류"]


@asynccontextmanager
async def lifespan(app):
    saved = joblib.load(CLF_PATH)                       # 분류기, 임베딩 모델 이름, 판
    state["clf"], state["판"] = saved["분류기"], saved["판"]
    state["emb"] = SentenceTransformer(saved["임베딩"], device="cpu")
    try:
        ask_llm("데우기 문의입니다", timeout=300)         # LLM 도 미리 올려 둔다
    except (OSError, ValueError, KeyError):            # 데우기에 실패해도 서버는 뜬다(분류기는 쓸 수 있다)
        state["데우기실패"] = True
    state["준비"] = True
    yield
    state.clear()


app = FastAPI(title="고객 문의 분류 API", version="1.0", lifespan=lifespan)


def classify_many(texts):
    """임베딩은 한 번에 계산하고, 확신도가 낮은 문의만 LLM 에 넘긴다. LLM 이 두 번 실패하면 분류기의 답을 쓴다"""
    X = state["emb"].encode(["query: " + t for t in texts], normalize_embeddings=True)
    out = []
    for text, p in zip(texts, state["clf"].predict_proba(X)):
        label, conf, by = str(state["clf"].classes_[p.argmax()]), float(p.max()), "분류기"
        if conf < THRESHOLD:
            for _ in range(2):
                try:
                    label, by = ask_llm(text), "LLM"
                    state["LLM호출"] += 1
                    break
                except (OSError, ValueError, KeyError):
                    state["LLM실패"] += 1
                    by = "분류기(LLM 실패)"
                    post("/api/generate", {"model": MODEL, "keep_alive": 0}, 30)
        out.append({"문의": text, "분류": label, "확신도": round(conf, 3), "판단": by})
    return out


Text = Annotated[str, Field(min_length=2, max_length=300)]      # 문의 하나는 2~300자


class One(BaseModel):
    문의: Text


class Many(BaseModel):
    문의들: list[Text] = Field(min_length=1, max_length=50)       # 한 번에 1~50개


@app.get("/health")
def health():
    return {"준비": state.get("준비", False), "판": state.get("판"), "기준값": THRESHOLD, "모델": MODEL, "데우기실패": state.get("데우기실패"),
            "LLM호출": state.get("LLM호출"), "LLM실패": state.get("LLM실패")}


@app.post("/classify")
def classify(req: One):
    t = time.perf_counter()
    return {**classify_many([req.문의])[0], "판": state["판"], "걸린초": round(time.perf_counter() - t, 3)}


@app.post("/classify/batch")
def classify_batch(req: Many):
    t = time.perf_counter()
    results = classify_many(req.문의들)
    return {"결과": results, "LLM호출": sum(r["판단"] == "LLM" for r in results), "판": state["판"],
            "걸린초": round(time.perf_counter() - t, 3)}
'''
from pathlib import Path
Path("out_inquiry_api.py").write_text(INQUIRY_API.lstrip(), encoding="utf-8")
print("out_inquiry_api.py ·", len(INQUIRY_API.splitlines()), "줄")

sys.path.insert(0, ".")
from fastapi.testclient import TestClient
import out_inquiry_api as api

with TestClient(api.app) as client:
    print("health:", client.get("/health").json())
    one = client.post("/classify", json={"문의": "주문한 지 나흘이 지났는데 아직 안 왔어요."}).json()
    print("하나:", one)
    batch = client.post("/classify/batch", json={"문의들": test["문의"].tolist()}).json()
    got = np.array([r["분류"] for r in batch["결과"]])
    print(f"묶음 40개: 정확도 {(got == test['분류']).mean():.3f} · LLM 호출 {batch['LLM호출']} · {batch['걸린초']}초 · 판 {batch['판']}")
    print("판단별:", pd.Series([r["판단"] for r in batch["결과"]]).value_counts().to_dict())
    print("빈 글:", client.post("/classify", json={"문의": ""}).status_code,
          "· 51개:", client.post("/classify/batch", json={"문의들": ["배송 언제 와요?"] * 51}).status_code,
          "· 빈 목록:", client.post("/classify/batch", json={"문의들": []}).status_code)
    for r, y in zip(batch["결과"], test["분류"]):
        if r["분류"] != y:
            print(f"  틀림: {r['문의']} 정답 {y} · {r['분류']}({r['판단']}, 확신도 {r['확신도']})")