실습 3 — Ollama 의 양자화 단계 견주기

import json, re, time, urllib.request
import numpy as np
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")][:12]
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")     # GPU 는 Ollama 에 비워 둔다
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]    # 모든 모델에 같은 조각을 준다
print(len(articles), "조항 ·", len(qa), "질문")

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=300).read())


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


SYSTEM = "당신은 솔내기술(가상 회사)의 규정 도우미입니다. [조항]에 있는 내용만으로 한 문장으로 답하세요."


BROKEN = re.compile(r"(.)\1{9,}")             # 같은 글자가 10번 넘게 이어지면 깨진 답으로 본다(예: 8888…)
retried = {"수": 0}


def ask(model, context, question):
    """답과 함께 Ollama 가 재 준 생성 토큰 수(eval_count)와 생성 시간(eval_duration, 나노초)을 돌려준다.
    깨진 답이 나오면 모델을 내렸다가 두 번까지 다시 묻는다(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["eval_count"], r["eval_duration"] / 1e9, r.get("load_duration", 0) / 1e9


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


unload_all()
print(ask("qwen2.5:0.5b", contexts[0], qa[0]["질문"]))

tags = {"Q4_K_M": "qwen2.5:0.5b", "Q8_0": "qwen2.5:0.5b-instruct-q8_0", "F16": "qwen2.5:0.5b-instruct-fp16"}
sizes = {m["name"]: m["size"] for m in json.loads(urllib.request.urlopen("http://localhost:11434/api/tags").read())["models"]}
rows, answers = [], {}
for level, model in tags.items():
    unload_all()
    _, _, _, load_sec = ask(model, contexts[0], qa[0]["질문"])               # 처음 한 번: 모델을 올린다
    vram = {m["name"]: m for m in json.loads(urllib.request.urlopen("http://localhost:11434/api/ps").read())["models"]}[model]
    right, tokens, secs = 0, 0, 0.0
    retried["수"] = 0
    for x, ctx in zip(qa, contexts):
        a, n, s, _ = ask(model, ctx, x["질문"])
        right += correct(a, x["정답열쇠"])
        tokens, secs = tokens + n, secs + s
        answers[(level, x["질문"])] = a
    rows.append({"단계": level, "파일MB": round(sizes[model] / 1e6), "GPU MB": round(vram["size_vram"] / 1e6),
                 "올리기(초)": round(load_sec, 1), "초당토큰": round(tokens / secs, 1), "정답": f"{right}/{len(qa)}", "다시 물음": retried["수"]})
import pandas as pd
print(pd.DataFrame(rows).to_string(index=False))
unload_all()

for x in qa:
    got = {lv: correct(answers[(lv, x["질문"])], x["정답열쇠"]) for lv in tags}
    if len(set(got.values())) > 1:                       # 단계마다 맞고 틀림이 다른 질문만
        print(x["질문"], f"(열쇠 {x['정답열쇠']})")
        for lv in tags:
            print(f"   {lv} {'O' if got[lv] else 'X'}: {answers[(lv, x['질문'])][:60]}")