실습 6 — 조각 수와 답 길이 줄이기

import json, re, 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("|"))


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")]
SYSTEM = "당신은 솔내기술(가상 회사)의 규정 도우미입니다. [조항]에 있는 내용만으로 답하세요."
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
ce = CrossEncoder("cross-encoder/mmarco-mMiniLMv2-L12-H384-v1", device="cpu")
A = emb.encode(["passage: " + a for a in articles], normalize_embeddings=True)

ranked = []                                              # 질문마다 재정렬한 조항 번호(앞이 더 가깝다)
for x in qa:
    qv = emb.encode(["query: " + x["질문"]], normalize_embeddings=True)[0]
    cand = np.argsort(-(A @ qv))[:10]
    ranked.append(cand[np.argsort(-ce.predict([(x["질문"], articles[i]) for i in cand]))])
unload_all()
chat([{"role": "user", "content": "안녕"}], 5)            # 몸풀기
print(len(ranked), "개 질문의 검색 순위 준비")

STYLES = {"한 문장": ("한 문장으로 답하세요.", 80), "자세히": ("자세히 설명해 주세요.", 300)}


def evaluate(k, style):
    """조각 k 개와 답의 꼴(style)로 질문 25개를 묻고 평균 입력·출력 토큰, 평균 생성 시간, 정답 수를 잰다"""
    instr, n_pred = STYLES[style]
    n_in, n_out, secs, right, answers = [], [], [], 0, []
    for x, cand in zip(qa, ranked):
        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{x['질문']}\n{instr}"}], num_predict=n_pred)
        a = r["message"]["content"].strip()
        n_in.append(r["prompt_eval_count"]); n_out.append(r["eval_count"])
        secs.append(r["total_duration"] / 1e9)              # Ollama 가 잰 시간(나노초 → 초)
        right += correct(a, x["정답열쇠"]); answers.append(a)
    return {"조각 수": k, "답": style, "입력토큰": round(np.mean(n_in)), "출력토큰": round(np.mean(n_out)),
            "생성 평균(초)": round(float(np.mean(secs)), 2), "정답": right}, answers


row, _ = evaluate(3, "한 문장")
# 결과: 조각 3개, 한 문장 답의 평균과 정답 수
print(row)

rows, answers = [], {}
for k, style in [(1, "한 문장"), (3, "한 문장"), (5, "한 문장"), (8, "한 문장"), (3, "자세히"), (8, "자세히")]:
    row, ans = evaluate(k, style)
    rows.append(row); answers[(k, style)] = ans
table = pd.DataFrame(rows)
# 결과: 설정마다 입력·출력 토큰, 생성 시간, 정답 수
print(table.to_string(index=False))
print("다시 물음", retried["수"])

for i, x in enumerate(qa):
    a1, a3 = answers[(1, "한 문장")][i], answers[(3, "한 문장")][i]
    if not correct(a1, x["정답열쇠"]) and correct(a3, x["정답열쇠"]):   # 조각 1개로는 틀리고 3개로는 맞힌 질문
        top = [articles[j][:articles[j].index(")") + 1] for j in ranked[i][:3]]
        print(f"{x['질문']} (근거 {x['근거조항']}) · 순위 1~3: {top}")
        print("   k=1:", a1[:70])
        print("   k=3:", a3[:70])