직접 해 보기: 작은 RAG 만들기 (BM25 + 벡터 + RRF + 평가)

12345678910111213flowchart LR
  D[문서 12개] --> W[단어 토큰<br/>BM25 색인]
  D --> G[글자 2-그램<br/>TF-IDF 벡터 색인]
  Q[질문] --> BM[BM25 점수]
  Q --> VS[코사인 유사도]
  W --> BM
  G --> VS
  BM --> H1[점수 0 초과만 순위 목록]
  VS --> H2[점수 0 초과만 순위 목록]
  H1 --> R[RRF k=60]
  H2 --> R
  R --> P[상위 2개로 프롬프트 조립<br/>모의 생성]
  R --> E[평가: recall@1 · @3 · MRR]

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116# 작은 RAG: BM25 + TF-IDF 벡터 + RRF + 평가 (표준 라이브러리 + numpy)
import math, re
from collections import Counter
import sys
import numpy as np
sys.stdout.reconfigure(encoding="utf-8")

DOCS = [
    "연차 휴가는 입사 1년이 지나면 15일이 주어지고, 남은 연차는 다음 해로 이월할 수 없습니다.",
    "재택근무는 주 2회까지 가능하며, 전날 오후 6시까지 팀장에게 신청해야 합니다.",
    "출장비는 출장 종료 후 7일 안에 영수증을 첨부해 경비 시스템으로 청구합니다.",
    "사내 와이파이 비밀번호는 매월 1일에 바뀌며, 보안팀 공지 게시판에서 확인할 수 있습니다.",
    "노트북이 고장 나면 IT 지원 포털에 접수하고, 대체 장비는 보통 이틀 안에 지급됩니다.",
    "오류 코드 ERR-4012는 VPN 인증서가 만료되었다는 뜻이며, 인증서를 다시 발급받으면 해결됩니다.",
    "오류 코드 ERR-4021은 비밀번호를 5회 틀려 계정이 잠긴 상태로, 30분 뒤 자동으로 풀립니다.",
    "점심 식대는 하루 1만 원까지 법인카드로 결제할 수 있습니다.",
    "신규 입사자는 첫 주에 보안 교육과 개인정보 보호 교육을 반드시 이수해야 합니다.",
    "회의실 예약은 그룹웨어 캘린더에서 하며, 최대 2시간까지 잡을 수 있습니다.",
    "육아휴직은 자녀 1명당 최대 1년까지 쓸 수 있고, 인사팀에 30일 전까지 신청합니다.",
    "택배는 1층 안내 데스크에서 받을 수 있으며, 3일이 지나면 반송됩니다.",
]
EVAL = [  # (질문, 정답 문서 번호)
    ("집에서 일하려면 언제까지 신청해야 하나요?", 1),
    ("ERR-4012 오류는 어떻게 해결하나요?", 5),
    ("남은 휴가를 내년에 쓸 수 있나요?", 0),
    ("출장 다녀와서 돈은 어떻게 돌려받나요?", 2),
    ("와이파이 비밀번호가 바뀌었어요", 3),
    ("계정이 잠겼을 때 얼마나 기다려야 하나요?", 6),
    ("노트북 고장 시 대체 장비는 언제 받나요?", 4),
    ("아이가 태어나면 휴직을 얼마나 쓸 수 있나요?", 10),
    ("ERR-4021이 떴어요", 6),
    ("연차는 며칠 받나요?", 0),
    ("점심값은 얼마까지 회사가 내 주나요?", 7),
    ("VPN 접속이 안 돼요", 5),
    ("회의실은 몇 시간까지 잡을 수 있나요?", 9),
    ("아기 돌보려고 한동안 쉬고 싶어요", 10),
]

def words(text):
    # 키워드 검색용: 공백 · 문장부호로 자른 단어(조사가 붙은 채로)
    return [w.lower() for w in re.findall(r"[0-9A-Za-z\-]+|[가-힣]+", text)]

def bigrams(text):
    # 벡터용: 단어 안의 글자 2-그램(형태가 조금 달라도 겹치게)
    out = []
    for w in words(text):
        out += [w[i:i+2] for i in range(len(w) - 1)] or [w]
    return out

DT = [words(d) for d in DOCS]      # BM25 색인
DB = [bigrams(d) for d in DOCS]    # 벡터 색인
N = len(DOCS); avgdl = sum(map(len, DT)) / N
df = Counter(t for d in DT for t in set(d))      # 단어 문서 빈도
dfb = Counter(t for d in DB for t in set(d))     # 2-그램 문서 빈도

def bm25_scores(q, k1=1.5, b=0.75):
    qt = words(q); s = np.zeros(N)
    for i, d in enumerate(DT):
        tf = Counter(d)
        for t in qt:
            if t not in tf: continue
            idf = math.log(1 + (N - df[t] + 0.5) / (df[t] + 0.5))
            s[i] += idf * tf[t] * (k1 + 1) / (tf[t] + k1 * (1 - b + b * len(d) / avgdl))
    return s

VOCAB = {t: j for j, t in enumerate(dfb)}
def tfidf_vec(toks):
    v = np.zeros(len(VOCAB))
    for t, c in Counter(toks).items():
        if t in VOCAB: v[VOCAB[t]] = (1 + math.log(c)) * math.log(N / dfb[t] + 1)
    n = np.linalg.norm(v); return v / n if n else v
DV = np.stack([tfidf_vec(d) for d in DB])
def vec_scores(q): return DV @ tfidf_vec(bigrams(q))   # 코사인 유사도

def rank(scores): return list(np.argsort(-scores, kind="stable"))
def hits(scores): return [i for i in rank(scores) if scores[i] > 0]   # 점수 0인 문서는 '검색 안 됨'으로
def rrf(*rankings, k=60):
    s = np.zeros(N)
    for r in rankings:
        for pos, i in enumerate(r, 1): s[i] += 1 / (k + pos)
    return rank(s)

def evaluate(name, fn):
    r1 = r3 = mrr = 0
    for q, gold in EVAL:
        r = fn(q); p = r.index(gold) + 1
        r1 += p <= 1; r3 += p <= 3; mrr += 1 / p
    n = len(EVAL)
    print(f"{name:8s} recall@1={r1/n:.3f} recall@3={r3/n:.3f} MRR={mrr/n:.3f}")

methods = {
    "BM25": lambda q: rank(bm25_scores(q)),
    "TF-IDF": lambda q: rank(vec_scores(q)),
    "RRF": lambda q: rrf(hits(bm25_scores(q)), hits(vec_scores(q))),
}
for name, fn in methods.items(): evaluate(name, fn)

print("\n질문별 정답 문서 순위 (BM25 / TF-IDF / RRF)")
for q, gold in EVAL:
    print(f"- {q} → " + " / ".join(str(fn(q).index(gold) + 1) for fn in methods.values()))

def answer(q, top_k=2):
    # 생성 단계는 LLM 대신 템플릿(모의)으로: 프롬프트만 조립해 보여 줌
    ids = methods["RRF"](q)[:top_k]
    ctx = "\n".join(f"[{j+1}] {DOCS[i]}" for j, i in enumerate(ids))
    return f"다음 자료만 근거로 답하고, 근거 번호를 붙이세요. 자료에 없으면 '자료에 없음'이라고 답하세요.\n{ctx}\n질문: {q}"
print("\n--- 조립된 프롬프트 예 ---")
print(answer("ERR-4012 오류는 어떻게 해결하나요?"))

# 쿼리 재작성(모의): 실제로는 LLM이 쓰는 검색용 질문을 사람이 대신 적어 둔 것
REWRITE = {"아기 돌보려고 한동안 쉬고 싶어요": "육아휴직은 최대 얼마나 쓸 수 있나",
           "점심값은 얼마까지 회사가 내 주나요?": "점심 식대 한도 법인카드"}
print("\n--- 쿼리 재작성 전후 정답 순위 (RRF) ---")
for q, new in REWRITE.items():
    gold = dict(EVAL)[q]
    print(f"- {q} → {methods['RRF'](q).index(gold)+1}위 / '{new}' → {methods['RRF'](new).index(gold)+1}위")

1234567891011121314151617181920212223242526272829BM25     recall@1=0.786 recall@3=0.857 MRR=0.837
TF-IDF   recall@1=0.929 recall@3=0.929 MRR=0.935
RRF      recall@1=0.929 recall@3=0.929 MRR=0.935

질문별 정답 문서 순위 (BM25 / TF-IDF / RRF)
- 집에서 일하려면 언제까지 신청해야 하나요? → 1 / 1 / 1
- ERR-4012 오류는 어떻게 해결하나요? → 1 / 1 / 1
- 남은 휴가를 내년에 쓸 수 있나요? → 2 / 1 / 1
- 출장 다녀와서 돈은 어떻게 돌려받나요? → 1 / 1 / 1
- 와이파이 비밀번호가 바뀌었어요 → 1 / 1 / 1
- 계정이 잠겼을 때 얼마나 기다려야 하나요? → 1 / 1 / 1
- 노트북 고장 시 대체 장비는 언제 받나요? → 1 / 1 / 1
- 아이가 태어나면 휴직을 얼마나 쓸 수 있나요? → 1 / 1 / 1
- ERR-4021이 떴어요 → 1 / 1 / 1
- 연차는 며칠 받나요? → 1 / 1 / 1
- 점심값은 얼마까지 회사가 내 주나요? → 8 / 1 / 1
- VPN 접속이 안 돼요 → 1 / 1 / 1
- 회의실은 몇 시간까지 잡을 수 있나요? → 1 / 1 / 1
- 아기 돌보려고 한동안 쉬고 싶어요 → 11 / 11 / 11

--- 조립된 프롬프트 예 ---
다음 자료만 근거로 답하고, 근거 번호를 붙이세요. 자료에 없으면 '자료에 없음'이라고 답하세요.
[1] 오류 코드 ERR-4012는 VPN 인증서가 만료되었다는 뜻이며, 인증서를 다시 발급받으면 해결됩니다.
[2] 오류 코드 ERR-4021은 비밀번호를 5회 틀려 계정이 잠긴 상태로, 30분 뒤 자동으로 풀립니다.
질문: ERR-4012 오류는 어떻게 해결하나요?

--- 쿼리 재작성 전후 정답 순위 (RRF) ---
- 아기 돌보려고 한동안 쉬고 싶어요 → 11위 / '육아휴직은 최대 얼마나 쓸 수 있나' → 1위
- 점심값은 얼마까지 회사가 내 주나요? → 1위 / '점심 식대 한도 법인카드' → 1위