실습 6 — 하이브리드와 재정렬

import json, time
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

text = open("it_helpdesk.md", encoding="utf-8").read()
chunks = ["## " + s.strip() for s in text.split("\n## ")[1:]]
qa = [json.loads(line) for line in open("it_helpdesk_qa.jsonl", encoding="utf-8")]
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
E = emb.encode(["passage: " + c for c in chunks], normalize_embeddings=True)
Q = emb.encode(["query: " + r["질문"] for r in qa], normalize_embeddings=True)
vec = TfidfVectorizer(analyzer="char_wb", ngram_range=(2, 3))
X = vec.fit_transform(chunks)
SEM = Q @ E.T                                                        # 의미 점수 (36, 30)
KW = cosine_similarity(vec.transform([r["질문"] for r in qa]), X)    # 키워드 점수 (36, 30)
GOLD = [[i for i, c in enumerate(chunks) if r["근거"] in c] for r in qa]

def evaluate(orders):
    """질문마다의 조각 순서 목록을 받아 hit@1, hit@3, MRR"""
    ranks = np.array([min(o.index(g) for g in gold) + 1 for o, gold in zip(orders, GOLD)])
    return {"hit@1": int((ranks == 1).sum()), "hit@3": int((ranks <= 3).sum()), "MRR": round(float((1 / ranks).mean()), 3)}

def order_of(scores):
    return np.argsort(-scores).tolist()

print("키워드:", evaluate([order_of(s) for s in KW]))
print("의미  :", evaluate([order_of(s) for s in SEM]))

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

rows = []
for alpha in [0.0, 0.3, 0.5, 0.7, 0.8, 1.0]:
    # α 는 의미 점수의 무게, 1 - α 는 키워드 점수의 무게
    orders = [order_of(alpha * minmax(s) + (1 - alpha) * minmax(k)) for s, k in zip(SEM, KW)]
    rows.append({"α": alpha, **evaluate(orders)})
print(pd.DataFrame(rows).to_string(index=False))

def rrf(order_a, order_b, k=60):
    """두 순서에서 순위의 역수 1/(k + 순위)를 더해 다시 줄 세운다"""
    score = {}
    for order in (order_a, order_b):
        for rank, i in enumerate(order, start=1):
            score[i] = score.get(i, 0) + 1 / (k + rank)
    return sorted(score, key=lambda i: -score[i])

rrf_orders = [rrf(order_of(k), order_of(s)) for s, k in zip(SEM, KW)]
print("RRF:", evaluate(rrf_orders))

ce = CrossEncoder("cross-encoder/mmarco-mMiniLMv2-L12-H384-v1", device="cpu")
HYB = [order_of(0.7 * minmax(s) + 0.3 * minmax(k)) for s, k in zip(SEM, KW)]   # 2단계에서 고른 α = 0.7

start = time.time()
reranked = []
for r, order in zip(qa, HYB):
    cand = order[:10]                                         # 1단계: 하이브리드 위 10개
    ce_scores = ce.predict([(r["질문"], chunks[i]) for i in cand])
    new = [cand[i] for i in np.argsort(-ce_scores)]          # 2단계: 크로스 인코더 점수로 다시 줄 세운다
    reranked.append(new + order[10:])
print(f"재정렬 36문항 {time.time() - start:.1f}초")
print("하이브리드(α=0.7):", evaluate(HYB))
print("+ 재정렬         :", evaluate(reranked))