프로젝트 1 — 사내 규정 검색기와 모델 비교표

import json, re, time, urllib.request
import numpy as np
import pandas as pd
from sentence_transformers import SentenceTransformer, CrossEncoder
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

pieces, chapter = [], ""
for line in open("company_rules.txt", encoding="utf-8").read().split("\n"):
    if re.match(r"^제\d+장 ", line):
        chapter = line
    elif line.startswith("제") and "조(" in line:
        m = re.match(r"제(\d+)조\((.+?)\)", line)
        pieces.append({"조": int(m.group(1)), "제목": m.group(2), "글": chapter + " " + line})   # 장 제목을 앞에 붙인다
qa = [json.loads(line) for line in open("company_rules_qa.jsonl", encoding="utf-8")]
title_to_no = {p["제목"]: p["조"] for p in pieces}
docs = [p["글"] for p in pieces]

# 임베딩과 재정렬 모델은 CPU 에 둔다. GPU 는 3단계의 Ollama 모델에 비워 둔다
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
ce = CrossEncoder("cross-encoder/mmarco-mMiniLMv2-L12-H384-v1", device="cpu")
E = emb.encode(["passage: " + d for d in docs], normalize_embeddings=True)
vec = TfidfVectorizer(analyzer="char_wb", ngram_range=(2, 3))
X = vec.fit_transform(docs)
print("조각", len(docs), "개 ·", docs[4])

def minmax(a):
    return (a - a.min()) / (a.max() - a.min())

def search(question, k=3, rerank=True):
    """가중 합(α=0.7) 하이브리드로 위 10개를 찾고, rerank 면 크로스 인코더로 다시 줄 세운다"""
    sem = E @ emb.encode("query: " + question, normalize_embeddings=True)
    kw = cosine_similarity(vec.transform([question]), X)[0]
    cand = np.argsort(-(0.7 * minmax(sem) + 0.3 * minmax(kw)))[:10].tolist()
    if rerank:
        s = ce.predict([(question, docs[i]) for i in cand])
        cand = [cand[i] for i in np.argsort(-s)]
    return cand[:k]

inside = [r for r in qa if r["근거조항"]]
for rr in [False, True]:
    found = [[pieces[i]["조"] for i in search(r["질문"], rerank=rr)] for r in inside]
    gold = [title_to_no[r["근거조항"]] for r in inside]
    h1 = sum(f[0] == g for f, g in zip(found, gold))
    h3 = sum(g in f for f, g in zip(found, gold))
    print("재정렬" if rr else "하이브리드", f"hit@1 {h1}/20 · hit@3 {h3}/20")

from openai import OpenAI, APIError

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
SYSTEM = ("당신은 솔내기술 인사팀의 안내 직원입니다. 아래 [규정] 조항을 읽고 질문에 한두 문장으로 답하세요. "
          "질문과 관련된 조항이 있으면 그 조항의 내용으로 답하고, 답 끝에 (제N조) 꼴로 근거를 적으세요. "
          "관련된 조항이 하나도 없을 때만 '규정에서 찾지 못했습니다.'라고 답하세요.")
REFUSE = ["찾지 못", "찾을 수 없", "나와 있지 않", "제공되지 않"]
# 검색은 한 번만 하고, 네 모델에 똑같은 프롬프트를 넣는다
prompts = [[{"role": "system", "content": SYSTEM},
            {"role": "user", "content": "[규정]\n" + "\n".join(docs[i] for i in search(r["질문"])) + "\n\n[질문]\n" + r["질문"]}]
           for r in qa]

def unload(model):
    """Ollama 에 keep_alive=0 을 보내 모델을 GPU 에서 바로 내린다"""
    body = json.dumps({"model": model, "keep_alive": 0}).encode()
    urllib.request.urlopen(urllib.request.Request("http://localhost:11434/api/generate", data=body,
                                                  headers={"Content-Type": "application/json"})).read()

def ask(model, messages):
    for _ in range(3):
        try:
            resp = client.chat.completions.create(model=model, messages=messages, temperature=0, seed=42, max_tokens=150)
            return resp.choices[0].message.content.strip()
        except APIError as e:
            error = e.message[:40]
            unload(model)              # 오류가 나면 모델을 내렸다가(다음 요청 때 새로 올라온다) 다시 보낸다
            time.sleep(2)
    return f"[오류] {error}"

MODELS = ["qwen2.5:1.5b", "qwen2.5:3b", "gemma3:1b", "llama3.2:1b"]
answers, seconds = {}, {}
for m in MODELS:
    ask(m, [{"role": "user", "content": "안녕하세요"}])   # 모델을 먼저 올려 둔다
    start = time.time()
    answers[m] = [ask(m, p) for p in prompts]
    seconds[m] = round(time.time() - start, 1)
    unload(m)                                            # 다음 모델을 위해 GPU 를 비운다
    print(m, seconds[m], "초 · 오류", sum(a.startswith("[오류]") for a in answers[m]), "개")

LICENSE = {"qwen2.5:1.5b": "Apache 2.0", "qwen2.5:3b": "Qwen Research",
           "gemma3:1b": "Gemma 이용 약관", "llama3.2:1b": "Llama 3.2 커뮤니티"}
rows = []
for m in MODELS:
    c = {"정답(/20)": 0, "출처 맞음(/20)": 0, "지나친 거절": 0, "모름 맞음(/5)": 0}
    for r, a in zip(qa, answers[m]):
        refused = any(w in a for w in REFUSE)
        if r["근거조항"]:
            c["정답(/20)"] += any(k in a for k in r["정답열쇠"].split("|"))
            c["출처 맞음(/20)"] += str(title_to_no[r["근거조항"]]) in re.findall(r"제\s*(\d+)\s*조", a)
            c["지나친 거절"] += refused
        else:
            c["모름 맞음(/5)"] += refused
    rows.append({"모델": m, **c, "25문항 초": seconds[m], "라이선스": LICENSE[m]})
print(pd.DataFrame(rows).to_string(index=False))

# 규정에 없는 질문 하나와 있는 질문 하나의 네 답을 읽어 본다
for q in ["명절 선물은 무엇을 주나요?", "부산 출장 가면 호텔비는 얼마까지 나와요?"]:
    n = next(i for i, r in enumerate(qa) if r["질문"] == q)
    print("질문:", q)
    for m in MODELS:
        print(f"  {m:13s}", answers[m][n].replace("\n", " ")[:80])