프로젝트 2 — 가전제품 설명서 Q&A

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

text = open("washer_manual.md", encoding="utf-8").read()
sections = [s.strip() for s in text.split("\n## ")[1:]]       # 절 16개. 첫 줄이 제목
qa = [json.loads(l) for l in open("washer_qa.jsonl", encoding="utf-8")]

def table_whole(tag):
    """절 하나를 조각 하나로. tag(제목) 는 조각 맨 앞에 붙일 글을 만드는 함수"""
    return [(tag(s.split("\n")[0]) + " " + s.split("\n", 1)[1], s.split("\n")[0]) for s in sections]

def table_rows(tag):
    """표가 든 절은 행마다 '머리 칸 이름 + 값' 문장 하나로 바꾼다"""
    out = []
    for s in sections:
        title, body = s.split("\n", 1)
        if body.startswith("|"):
            lines = body.split("\n")
            head = [c.strip() for c in lines[0].strip("|").split("|")]
            for line in lines[1:]:
                cells = [c.strip() for c in line.strip("|").split("|")]
                out.append((tag(title) + " " + ", ".join(f"{h} {c}" for h, c in zip(head, cells)) + ".", title))
        else:
            out.append((tag(title) + " " + body, title))
    return out

hash_tag = lambda title: f"## {title}\n"          # 판 1: 마크다운 제목 꼴
bracket_tag = lambda title: f"[절: {title}]"        # 판 2: 답에 적을 출처 꼴 그대로
print("표 통째로:", len(table_whole(hash_tag)), "조각 · 표를 행으로:", len(table_rows(hash_tag)), "조각")
for c, title in table_rows(bracket_tag)[18:21]:
    print(" ", c[:90])

emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")

class Search:
    def __init__(self, docs):
        self.E = emb.encode(["passage: " + d for d in docs], normalize_embeddings=True)
        self.vec = TfidfVectorizer(analyzer="char_wb", ngram_range=(2, 3))
        self.X = self.vec.fit_transform(docs)
    def top(self, question, k=3, alpha=0.7):
        sem = self.E @ emb.encode("query: " + question, normalize_embeddings=True)
        kw = cosine_similarity(self.vec.transform([question]), self.X)[0]
        mm = lambda a: (a - a.min()) / (a.max() - a.min() + 1e-9)
        return np.argsort(-(alpha * mm(sem) + (1 - alpha) * mm(kw)))[:k].tolist()

inside = [r for r in qa if r["근거"]]
rows = []
for way, make in [("표 통째로", table_whole), ("표를 행으로", table_rows)]:
    chunks = make(hash_tag)
    s = Search([c for c, _ in chunks])
    for name, alpha in [("키워드", 0.0), ("의미", 1.0), ("하이브리드", 0.7)]:
        hit = sum(r["근거"] in [chunks[i][1] for i in s.top(r["질문"], 3, alpha)] for r in inside)
        rows.append({"조각": way, "검색": name, "hit@3(/22)": hit})
print(pd.DataFrame(rows).pivot(index="조각", columns="검색", values="hit@3(/22)"))

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
MODEL = "qwen2.5:3b"
REFUSE = ["찾지 못", "찾을 수 없", "나와 있지 않", "제공되지 않"]
SYSTEM = ("당신은 포근 WD-210 세탁기의 고객 상담원입니다. 아래 [설명서] 내용만 근거로 한두 문장으로 답하세요. "
          "답 끝에 근거가 된 절 제목을 [절: 제목] 꼴로 적으세요. "
          "관련된 내용이 하나도 없을 때만 '설명서에서 찾지 못했습니다.'라고 답하세요.")

def unload(model):
    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(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}"

def evaluate(tag, show=0):
    """표를 행으로 바꾼 조각에 tag 꼴의 제목을 붙여 25문항을 묻고 센다"""
    chunks = table_rows(tag)
    s = Search([c for c, _ in chunks])
    count = {"정답(/22)": 0, "출처(/22)": 0, "지나친 거절": 0, "모름 맞음(/3)": 0}
    start = time.time()
    for n, r in enumerate(qa):
        context = "\n".join(chunks[i][0] for i in s.top(r["질문"]))
        a = ask([{"role": "system", "content": SYSTEM},
                 {"role": "user", "content": f"[설명서]\n{context}\n\n[질문]\n{r['질문']}"}])
        refused = any(w in a for w in REFUSE)
        if r["근거"]:
            count["정답(/22)"] += any(k in a for k in r["정답열쇠"].split("|"))
            count["출처(/22)"] += f"[절: {r['근거']}]" in a
            count["지나친 거절"] += refused
        else:
            count["모름 맞음(/3)"] += refused
        if n < show:
            print(" ", r["질문"], "→", a.replace("\n", " ")[:90])
    count["25문항 초"] = round(time.time() - start)
    return count

first = evaluate(hash_tag, show=6)
print("판 1(## 제목):", first)

second = evaluate(bracket_tag, show=6)
print("판 2([절: 제목]):", second)
print(pd.DataFrame([first, second], index=["판 1(## 제목)", "판 2([절: 제목])"]))