[RAG/팀] 1주차

---

RAG 기본 구조 만들기

사용자 쿼리 - 벡터DB 조회 - LLM

 

1. main.py가 아닌 build_index.py로 시작

# (main 파이프라인과 분리된 1회성 실행 스크립트)
# RRAG 핵심 프로세스가 아닌 사전 전처리 과정이기 때문
from backend.service.rag.ingestion.index_builder import IndexBuilder

# 전처리 단계: 문서를 임베딩하고 FAISS 인덱스를 빌드 후 저장
# 클래스 명인 IndexBuilder가 마음에 들지 않음.
builder = IndexBuilder()            # 자바로 따지면 IndexBuilder 클래스를 가지고 인스턴스 생성 후 builder라는 변수에 할당한 것.
embedder, store = builder.build()   # builder 객체의 build 메서드를 실행하여, build의 멤버 변수를 할당
store.save()                        # store 객체의 save 메서드를 실행하여 DB에 저장.

print("✅ FAISS index 저장 완료")

---

2. 

from pypdf import PdfReader
from backend.service.rag.components.embedding.embedder import Embedder
from backend.service.rag.ingestion.chunker import Chunker
from backend.service.rag.components.vectorstore.faiss.vector_store import VectorStore

# RAG ingestion 파이프라인을 하나의 객체로 캡슐화
# 외부에서는 build()만 호출하지만 실제로 내부에서는
# 문서 로딩 → 청킹 → 임베딩 → 벡터DB 저장 여러 단계가 실행됨.
# 복잡한 내부 과정을 숨기고 하나의 메서드로 제공한다 → 캡슐화
# ingestion = 데이터를 시스템에 넣는 과정
# 여기서는 문서를 벡터화하여 DB에 저장하는 작업이 됨.
class IndexBuilder:

    # 생성자, IndexBuilder 인스턴스가 생성되면 하위의 코드를 실행한다.
    # 작업 준비를 하는 것, 준비물 사놓기.
    def __init__(self):
        
        self.embedder = Embedder()                                      # 1. 임베딩을 처리할 모델을 준비한다.        
        self.chunker  = Chunker()                                       # 2. load 된 데이터를 청킹할 청커를 준비한다.
        self.vector_store = VectorStore()                               # 3. 처리된 문서 데이터를 저장할 벡터DB 준비한다.

    # build_index.py 파일에서 호출된 메서드가 호출되어 실제로 처리되는 곳
    # PDF 로딩 → 청킹 → 임베딩 → 벡터DB 저장 과정이 한 번에 실행되도록 만듦.
    def build_index(self):

        documents = self.load_documents()                               # 1. 처리할 문서를 임베딩 처리하기 위해 로딩한다.
        chunks = self.chunk_documents(documents)                        #.2. 청킹
        vectors = self.embed_chunks(chunks)                             #.3. 벡터 DB 저장
        self.vector_store.add_documents(vectors, chunks)
        return self.embedder, self.vector_store
    
    # 추후 Loader를 따로 파일 만들어야 겠다.
    def load_documents(self):

        reader = PdfReader("backend/data/raw/pdf/북브리프_돈의심리학.pdf")
        docs = []

        for page in reader.pages:
            text = page.extract_text()

            #print("text : ", text)
            if text:
                docs.append(text)
        return docs
    
    def chunk_documents(self, documents):

        chunks = []
        for page, doc in enumerate(documents):

            print(f"\n========= PAGE {page+1} =========")
            doc_chunks = self.chunker.split(doc, page+1)
            chunks.extend(doc_chunks)
        return chunks
    
    # 임베딩 모델은 문자열 리스트를 받아야 한다.
    def embed_chunks(self, chunks):
        
        # 문자열 리스트로 변환하는 코드 추가.
        texts = [c["text"] for c in chunks]
        return self.embedder.embed(texts)

---

개발하면서 오류가 났던 부분

이 방식은 .txt 파일에서만 사용한다.

with open("data/raw/pdf/북브리프_돈의심리학.pdf", "r", encoding="utf-8") as f:

 

이유1 : 파이썬은 현재 실행 위치 기준으로 경로를 잡는다.

내가 실행한건 맨 상위

C:\projects\rag\ai-architecture-examples

로 backend가 제외되었음.

그래서 FileNotFoundError 발생.

 

이유2 : 기존 코드에서 

open("북브리프_돈의심리학.pdf", "r")

 

이렇게 처리햐였는데, 이렇게 하면 PDF를 텍스트 파일처럼 읽으려고 함.

pdf는 이렇게 읽으면 안되고

보통 RAG에서는

from langchain.document_loaders import PyPDFLoader

이렇게 하지만 우리는 랭체인을 사용하지 않음으로

아래와 같이 pypdf를 설치

pip install pypdf

 

그리고 임포트하고 밑에서 코드로 이용한다.

from pypdf import PdfReader
reader = PdfReader("backend/data/raw/pdf/북브리프_돈의심리학.pdf")

위와 같이 PdfReader를 import 하고 이걸로 pdf 파일을 읽어오도록 코드를 수정함.

 

---

 

전처리시 신경쓰였던 것

 

이미지를 보면 알겠지만 [unk], [cls]가 함께 출력됨.

 

 

unk는 보통 PDF 텍스트에 특수문자 / 줄바꿈 / 인코딩 문제가 있을 때 생긴다.

’ ‘ “ ”

이런 문자, 이건 나중에 text 단계를 넣는게 좋다 해서 그리 처리하긴 했는데

'가 나와야 하는 경우가 있다면 어떻게 처리해야할지 다시 확인이 필요하다.

 

즉 PDF 자체 문제는 아님

해당 얘기 아직 이해못함

원인: tokenizer.decode() 가 unicode normalization을 안 해서
한글이 jamo 단위로 분리된 상태입니다.

 

[UNK] 많이 나오는 이유

이건 tokenizer vocabulary에 없는 문자입니다.

 

그래서 최종코드는?

import unicodedata

tokens = self.tokenizer.encode(text, add_special_tokens=False)

while start < len(tokens):

    end = start + self.chunk_size

    chunk_tokens = tokens[start:end]

    chunk_text = self.tokenizer.decode(
        chunk_tokens,
        skip_special_tokens=True,
        clean_up_tokenization_spaces=True
    )

    chunk_text = unicodedata.normalize("NFC", chunk_text)

    chunks.append(chunk_text)

    print("\n======================")
    print(chunk_text)
    print("======================")

    start += self.chunk_size - self.overlap

---

토큰 기준 청킹 VS 문자 기준 청킹

 

---