프로젝트 1 — 우리 GPU 에 맞는 모델 고르기

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

articles = [l.strip() for l in open("company_rules.txt", encoding="utf-8") if re.match(r"^제\d+조\(", l)]
qa = [json.loads(line) for line in open("company_rules_qa.jsonl", encoding="utf-8")]
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
A = emb.encode(["passage: " + a for a in articles], normalize_embeddings=True)
Q = emb.encode(["query: " + x["질문"] for x in qa], normalize_embeddings=True)
contexts = ["\n".join(articles[i] for i in np.argsort(-(A @ q))[:3]) for q in Q]
SYSTEM = "당신은 솔내기술(가상 회사)의 규정 도우미입니다. [조항]에 있는 내용만으로 한 문장으로 답하세요."


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


def loaded():
    return {m["name"]: m for m in json.loads(urllib.request.urlopen("http://localhost:11434/api/ps").read())["models"]}


def unload_all():
    for name in loaded():
        post("/api/generate", {"model": name, "keep_alive": 0})


BROKEN = re.compile(r"(.)\1{9,}")
retried = {"수": 0}


def ask(model, context, question):
    """깨진 답(같은 글자 10번 넘게)이면 모델을 내렸다가 두 번까지 다시 묻는다"""
    for _ in range(3):
        r = post("/api/chat", {"model": model, "stream": False, "options": {"temperature": 0, "seed": 42, "num_predict": 80},
                                "messages": [{"role": "system", "content": SYSTEM},
                                             {"role": "user", "content": f"[조항]\n{context}\n\n[질문]\n{question}"}]})
        answer = r["message"]["content"].strip()
        if not BROKEN.search(answer):
            break
        retried["수"] += 1
        post("/api/generate", {"model": model, "keep_alive": 0})
    return answer, r


correct = lambda a, key: any(k in a.replace(" ", "") for k in key.replace(" ", "").split("|"))
installed = {m["name"] for m in json.loads(urllib.request.urlopen("http://localhost:11434/api/tags").read())["models"]}
CANDIDATES = [c for c in ["qwen2.5:0.5b", "qwen2.5:0.5b-instruct-q8_0", "qwen2.5:0.5b-instruct-fp16", "gemma3:1b",
                          "llama3.2:1b", "qwen2.5:1.5b", "qwen2.5:3b"] if c in installed]
print(len(qa), "질문 · 후보", CANDIDATES)

rows, answers = [], {}
for model in CANDIDATES:
    unload_all()
    _, first = ask(model, contexts[0], qa[0]["질문"])                    # 올리기
    info = loaded()[model]
    right, secs, tokens, gen = 0, [], 0, 0.0
    retried["수"] = 0
    for x, ctx in zip(qa, contexts):
        a, r = ask(model, ctx, x["질문"])
        right += correct(a, x["정답열쇠"])
        secs.append(r["total_duration"] / 1e9)                         # 답 하나에 든 전체 시간
        tokens, gen = tokens + r["eval_count"], gen + r["eval_duration"] / 1e9
        answers[(model, x["질문"])] = a
    rows.append({"모델": model, "GPU 몫": round(info["size_vram"] / info["size"], 2), "메모리MB": round(info["size"] / 1e6),
                 "올리기(초)": round(first["load_duration"] / 1e9, 1), "평균(초)": round(float(np.mean(secs)), 2),
                 "초당토큰": round(tokens / gen, 1), "정답": right, "다시 물음": retried["수"]})
    print(rows[-1])
unload_all()

table = pd.DataFrame(rows)
table["조건"] = (table["GPU 몫"] >= 0.99) & (table["평균(초)"] <= 3.0)
print(table.to_string(index=False))
ok = table[table["조건"]].sort_values(["정답", "평균(초)"], ascending=[False, True])
choice = ok.iloc[0]
print(f"\n고른 모델: {choice['모델']} · 정답 {choice['정답']}/{len(qa)} · 평균 {choice['평균(초)']}초 · 메모리 {choice['메모리MB']}MB")
small = table[table["조건"] & (table["메모리MB"] <= 2000)].sort_values(["정답", "평균(초)"], ascending=[False, True])
print(f"메모리를 2GB 로 줄이면: {small.iloc[0]['모델']} · 정답 {small.iloc[0]['정답']}/{len(qa)}")

smallest = "qwen2.5:0.5b"
for x in qa:
    a_big, a_small = answers[(choice["모델"], x["질문"])], answers[(smallest, x["질문"])]
    if correct(a_big, x["정답열쇠"]) != correct(a_small, x["정답열쇠"]):
        print(x["질문"], f"(열쇠 {x['정답열쇠']})")
        print(f"   {choice['모델']}: {a_big[:60]}")
        print(f"   {smallest}: {a_small[:60]}")