실습 3 — 연습용 임베딩과 FAISS 로 검색 적중 재기

import re
import warnings
import zlib

warnings.filterwarnings("ignore", category=DeprecationWarning)

import pandas as pd
from langchain_community.document_loaders import PyPDFLoader
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter


class BigramEmbeddings(Embeddings):
    """글자 두 개 묶음(바이그램)을 세어 숫자 목록을 만드는 연습용 임베딩"""

    def __init__(self, size=512):
        self.size = size

    def _vector(self, text):
        text = text.replace(" ", "").replace("\n", "")
        vec = [0.0] * self.size
        for i in range(len(text) - 1):
            # 글자 두 개를 crc32 로 0~511 사이의 칸 번호로 바꿔 그 칸을 1 올린다
            vec[zlib.crc32(text[i:i + 2].encode("utf-8")) % self.size] += 1.0
        norm = sum(v * v for v in vec) ** 0.5 or 1.0
        return [v / norm for v in vec]

    def embed_documents(self, texts):
        return [self._vector(t) for t in texts]

    def embed_query(self, text):
        return self._vector(text)


emb = BigramEmbeddings()

# 전처리 전: 실습 2의 1단계와 같이 크기 200, 겹침 30 으로 나눈 조각
pages = PyPDFLoader("travel_terms.pdf").load()
raw_chunks = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=30).split_documents(pages)

# 전처리 후: 실습 2가 저장한 조항 단위 조각을 Document 로 되살린다
saved = pd.read_csv("out_ch06_chunks.csv")
art_chunks = [
    Document(page_content=row.내용, metadata={"article": row.article, "chapter": row.chapter, "page": int(row.page)})
    for row in saved.itertuples()
]
print("전처리 전 조각:", len(raw_chunks), "/ 전처리 후 조각:", len(art_chunks))
print("임베딩 숫자 개수:", len(emb.embed_query("환불은 언제 들어오나요?")))

# from_documents: 조각마다 embed_documents 로 숫자 목록을 만들어 FAISS 색인에 넣는다
raw_db = FAISS.from_documents(raw_chunks, emb)
art_db = FAISS.from_documents(art_chunks, emb)


def articles_of(doc):
    """조각에 든 조항 번호 목록. 조항 단위 조각은 metadata 에서, 크기로 자른 조각은 글 속의 조항 제목에서 찾는다"""
    if "article" in doc.metadata:
        return [doc.metadata["article"]]
    # (?= ): 바로 뒤에 빈칸이 오는 "제N조"만 센다. 본문 속 "제5조의 …" 같은 말은 빼려는 것이다
    return re.findall(r"제\d+조(?= )", doc.page_content)


def label(doc):
    """찍어 볼 이름표. 머리글이 든 조각에는 "머리글+"를 붙인다"""
    head = "머리글+" if "연습용 가상 문서" in doc.page_content else ""
    return head + ("·".join(articles_of(doc)) or "(조항 제목 없음)")


for q in ["출발 5일 전에 취소하면 수수료가 얼마예요?", "강아지 데려가도 돼요?"]:
    print("질문:", q)
    for name, db in [("전처리 전", raw_db), ("전처리 후", art_db)]:
        # similarity_search_with_score: (조각, 거리) 목록. 거리는 작을수록 가깝다
        top = db.similarity_search_with_score(q, k=2)
        print(f"  {name}:", [(label(d), round(float(s), 3)) for d, s in top])

questions = pd.read_csv("rag_questions.csv")


def hit(db, question, answer, k):
    """상위 k 개 조각 가운데 정답 조항(의 제목)이 든 조각이 있으면 True"""
    return any(answer in articles_of(d) for d in db.similarity_search(question, k=k))


rows = []
for row in questions.itertuples():
    top3 = [label(d) for d in art_db.similarity_search(row.질문, k=3)]
    rows.append({
        "번호": row.번호, "정답": row.정답조항,
        "전처리 전 1위": hit(raw_db, row.질문, row.정답조항, 1),
        "전처리 후 1위": hit(art_db, row.질문, row.정답조항, 1),
        "전처리 후 3위 안": hit(art_db, row.질문, row.정답조항, 3),
        "전처리 후 상위 3": " ".join(top3),
    })
hits = pd.DataFrame(rows)
print(hits.to_string(index=False))
print()
for col in ["전처리 전 1위", "전처리 후 1위", "전처리 후 3위 안"]:
    print(f"{col}: {int(hits[col].sum())} / {len(hits)}")

hits.to_csv("out_ch06_hits.csv", index=False, encoding="utf-8-sig")
print("저장: out_ch06_hits.csv")

from pathlib import Path

# save_local: 색인을 폴더에 저장한다. index.faiss(숫자 목록)와 index.pkl(조각의 글과 metadata)이 생긴다
art_db.save_local("out_ch06_faiss")
print("저장: out_ch06_faiss →", sorted(f.name for f in Path("out_ch06_faiss").iterdir()))

# load_local: 저장한 색인을 불러온다. 만들 때와 같은 임베딩을 넘겨야 질문을 같은 지도에 놓는다
# allow_dangerous_deserialization=True: index.pkl(pickle)을 열어도 된다고 허락한다. 내가 만든 색인에만 쓴다
loaded = FAISS.load_local("out_ch06_faiss", emb, allow_dangerous_deserialization=True)
print("불러온 조각 수:", loaded.index.ntotal)
doc, score = loaded.similarity_search_with_score("여권 기간이 얼마나 남아야 해요?", k=1)[0]
print("검색:", doc.metadata["article"], round(float(score), 3), "|", doc.page_content[:40], "…")