프로젝트 1 — 응답 시간 절반으로 줄이기

import json, re, time, urllib.request, urllib.error
import numpy as np
import pandas as pd
from sentence_transformers import SentenceTransformer, CrossEncoder

HOST = "http://127.0.0.1:11434"
MODEL = "qwen2.5:3b"
BROKEN = re.compile(r"(.)\1{9,}")
retried = {"수": 0}


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


def unload_all():
    for m in json.loads(urllib.request.urlopen(HOST + "/api/ps").read())["models"]:
        post("/api/generate", {"model": m["name"], "keep_alive": 0})


def chat(messages, num_predict):
    """실습 4와 같은 함수: 서버 오류나 깨진 답이면 모델을 내렸다 다시 묻는다"""
    for _ in range(3):
        try:
            r = post("/api/chat", {"model": MODEL, "stream": False, "messages": messages,
                                   "options": {"temperature": 0, "seed": 42, "num_predict": num_predict}})
            if not BROKEN.search(r["message"]["content"]):
                return r
        except urllib.error.HTTPError:
            pass
        retried["수"] += 1
        post("/api/generate", {"model": MODEL, "keep_alive": 0})
    return r


def correct(answer, key):
    return any(k in answer.replace(" ", "") for k in key.replace(" ", "").split("|"))


def normalize(q):
    return re.sub(r"\s+", " ", q).strip().rstrip("?.").strip()


articles = [l.strip() for l in open("company_rules.txt", encoding="utf-8") if re.match(r"^제\d+조\(", l)]
qa = [json.loads(l) for l in open("company_rules_qa.jsonl", encoding="utf-8")]
log = pd.read_csv("rag_query_log.csv")
SYSTEM = "당신은 솔내기술(가상 회사)의 규정 도우미입니다. [조항]에 있는 내용만으로 답하세요."
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
ce = CrossEncoder("cross-encoder/mmarco-mMiniLMv2-L12-H384-v1", device="cpu")


def slow_rag(q):
    """기준: 실습 4의 느린 RAG"""
    t = {}
    s = time.perf_counter()
    A = emb.encode(["passage: " + a for a in articles], normalize_embeddings=True)
    t["색인"] = time.perf_counter() - s; s = time.perf_counter()
    qv = emb.encode(["query: " + q], normalize_embeddings=True)[0]
    cand = np.argsort(-(A @ qv))
    t["검색"] = time.perf_counter() - s; s = time.perf_counter()
    cand = cand[np.argsort(-ce.predict([(q, articles[i]) for i in cand]))]
    t["재정렬"] = time.perf_counter() - s; s = time.perf_counter()
    ctx = "\n".join(articles[i] for i in cand[:8])
    r = chat([{"role": "system", "content": SYSTEM},
              {"role": "user", "content": f"[조항]\n{ctx}\n\n[질문]\n{q}\n자세히 설명해 주세요."}], num_predict=300)
    t["생성"] = time.perf_counter() - s
    return r["message"]["content"].strip(), t


# 고친 RAG 의 색인: 서버가 뜰 때 한 번만 만든다
A_INDEX = emb.encode(["passage: " + a for a in articles], normalize_embeddings=True)
K = 5                                                    # 실습 6에서 정답 수를 지킨 조각 수


def fast_rag(q):
    """고친 RAG: 색인 다시 쓰기, 재정렬 후보 10개, 조각 K 개, 한 문장 답"""
    t = {"색인": 0.0}
    s = time.perf_counter()
    qv = emb.encode(["query: " + q], normalize_embeddings=True)[0]
    cand = np.argsort(-(A_INDEX @ qv))[:10]              # 후보를 30개에서 10개로
    t["검색"] = time.perf_counter() - s; s = time.perf_counter()
    cand = cand[np.argsort(-ce.predict([(q, articles[i]) for i in cand]))]
    t["재정렬"] = time.perf_counter() - s; s = time.perf_counter()
    ctx = "\n".join(articles[i] for i in cand[:K])
    r = chat([{"role": "system", "content": SYSTEM},
              {"role": "user", "content": f"[조항]\n{ctx}\n\n[질문]\n{q}\n한 문장으로 답하세요."}], num_predict=80)
    t["생성"] = time.perf_counter() - s
    return r["message"]["content"].strip(), t


unload_all()
chat([{"role": "user", "content": "안녕"}], 5)             # 몸풀기
print("준비 끝 · 조항", len(articles), "· 질문", len(qa), "· 기록", len(log))

def measure(fn, label):
    """질문 25개를 fn 으로 묻고 단계별 평균, 평균·p50·p95 시간, 정답 수를 돌려준다"""
    rows = []
    for x in qa:
        a, t = fn(x["질문"])
        rows.append({**t, "전체": sum(t.values()), "정답": correct(a, x["정답열쇠"])})
    d = pd.DataFrame(rows)
    return {"RAG": label, **{k: round(d[k].mean(), 3) for k in ["색인", "검색", "재정렬", "생성"]},
            "평균(초)": round(d["전체"].mean(), 2), "p50(초)": round(d["전체"].median(), 2),
            "p95(초)": round(d["전체"].quantile(0.95), 2), "정답": int(d["정답"].sum())}


result = pd.DataFrame([measure(slow_rag, "기준(느린 RAG)"), measure(fast_rag, "고친 RAG")])
# 결과: 단계별 평균 시간, 전체의 평균·p50·p95, 정답 수
print(result.to_string(index=False))
ratio = result["평균(초)"][1] / result["평균(초)"][0]
print(f"평균 시간 비율 {ratio:.2f} · 정답 차이 {result['정답'][1] - result['정답'][0]} · 다시 물음 {retried['수']}")

cache, hits = {}, 0
start = time.perf_counter()
for q in log["질문"]:
    key = normalize(q)                                    # 띄어쓰기와 물음표를 다듬은 열쇠
    if key in cache:
        hits += 1
    else:
        cache[key] = fast_rag(q)[0]
cached_sec = time.perf_counter() - start
# 기준 RAG 로 80줄을 모두 처리하면 걸릴 시간은 평균 × 80 으로 어림한다(실제로 돌리면 몇 분 걸린다)
estimate = result["평균(초)"][0] * len(log)
print(f"고친 RAG + 답 캐시: {cached_sec:.1f}초 (캐시 맞힘 {hits}/{len(log)})")
print(f"기준 RAG 로 어림: {estimate:.0f}초 → {estimate / cached_sec:.1f}배 빠름")

base, fast = result.iloc[0], result.iloc[1]
saved = pd.DataFrame({"단계": ["색인", "검색", "재정렬", "생성"],
                      "기준(초)": [base[s] for s in ["색인", "검색", "재정렬", "생성"]],
                      "고친 뒤(초)": [fast[s] for s in ["색인", "검색", "재정렬", "생성"]]})
saved["줄어든 시간(초)"] = (saved["기준(초)"] - saved["고친 뒤(초)"]).round(3)
saved["줄어든 몫"] = (saved["줄어든 시간(초)"] / saved["줄어든 시간(초)"].sum()).round(3)
# 결과: 단계마다 줄어든 시간과, 전체 단축에서 차지하는 몫
print(saved.to_string(index=False))