LangChain을 이용한 RAG - (2) 문서 임베딩 편

2024.07.01 - [데이터 분석/LLM] - LangChain을 이용한 RAG - (1) 이론편

 

임베딩 방식

"문서를 어떻게 벡터화하지?"라는 근본적인 질문에는 Word2Vec, BERT 같은 임베딩 모델에 대해 직접 찾아보는 것을 추천한다.

여기서는 LangChain을 이용하여 임베딩을 수행하는 방법을 설명하고자 한다.

 

(1) 파일 불러오기
pdf, csv, txt, docx 파일, 데이터프레임 등 다양하게 불러올 수 있다. `DirectoryLoader`를 이용하여 여러 파일을 동시에 불러올 수도 있다.

from langchain.document_loaders import PyPDFLoader, CSVLoader, TextLoader, Docx2txtLoader, DataFrameLoader, DirectoryLoader

(2) 문서 분할
문서를 분할하는 데는 크게 두 가지 이유가 있다. 하나는 문서가 너무 길면 모델의 최대 토큰 수를 넘겨 작동하지 않게 되거나, 최대 토큰 수를 넘지 않았다 하더라도 비용이 많이 청구될 수 있기 때문이다. 다른 하나는 "정답"이 있는 문장만을 주었을 때 대답의 정확도가 가장 높다는 연구에 따라 프롬프트에 넣을 내용을 최대한 줄이고자 함이다. LangChain에는 `CharacterTextSplitter`, `RecursiveCharacterTextSplitter`, `TokenTextSplitter` 등이 있으며 대체로 `RecursiveCharacterTextSplitter`를 사용하는 듯하다.


(3) 임베딩
분할된 각 문서를 벡터화한다. 유료 모델인 `OpenAIEmbeddings`와 무료 모델인 `HuggingFaceBgeEmbeddings`, `FastEmbedEmbeddings` 등이 있다.

 

(1) 파일 불러오기

위에서 잠깐 소개했지만 pdf, csv 등 다양하게 파일을 불러올 수 있는데, 파일이 매우 많아서 불러오기가 귀찮다면 `DirectoryLoader`를 이용하여 여러 파일을 동시에 불러올 수 있다.

from langchain.document_loaders import DirectoryLoader

loader = DirectoryLoader(".", glob="data/*.txt", show_progress=False)
docs = loader.load()
print(f"문서의 수: {len(docs)}")

for i in range(len(docs)):
    print(f"\n[{i}번 문서 내용]\n{docs[i].page_content[800:900]}")

텍스트 파일 2개를 불러오는 데 13초가 걸렸다.

 

아래처럼 개별 데이터를 불러오면...

from langchain.document_loaders import TextLoader

files = ["data/Alice's Adventures in Wonderland.txt", "data/resell.txt"]

docs = []
for file in files:
    loader = TextLoader(file, encoding='utf8')
    docs.extend(loader.load())
print(f"문서의 수: {len(docs)}")

for i in range(len(docs)):
    print(f"\n[{i}번 문서 내용]\n{docs[i].page_content[800:900]}")

1.05ms가 걸렸다. 웬만하면 파일명을 지정해서 불러오도록 하자.

 

 

(2) 문서 분할

`CharacterTextSplitter`

from langchain.text_splitter import CharacterTextSplitter

text = docs[1].page_content

text_splitter = CharacterTextSplitter(
    separator="\n\n",
    chunk_size=50,
    chunk_overlap=10,
    length_function=len,
    is_separator_regex=False,
)

split_docs = text_splitter.split_text(text)

- 파라미터
    - `separator`: chunk를 나누는 기준이 된다.
    - `chunk_size`: 분리된 각 텍스트의 최대 chunk 길이
    - `chunk_overlap`: 겹치게 할 chunk의 길이
    - `length_function`: chunk의 길이를 측정할 함수
    - `is_seperator_regex`: `separator`를 정규식으로 볼 것인지 여부
- 작동 방식: 먼저 `separator`로 텍스트를 나눈 뒤 `chunk_size`를 초과하지 않도록 다시 합친다.

만약 문장을 아무리 쪼개도 `chunk_size`를 넘게 되면 아래처럼 경고가 뜬다.

 

`RecursiveCharacterTextSplitter`

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    separators = ["\n\n", "\n", " ", ""],
    chunk_size=100,
    chunk_overlap=10,
    length_function=len,
    is_separator_regex=False,
)

split_docs = text_splitter.split_text(text)

- 파라미터
    - `separators`: chunk를 나누는 기준 리스트. 큰 단위에서 작은 단위 순으로 작성한다.
    - `chunk_size`: 분리된 각 텍스트의 최대 chunk 길이
    - `chunk_overlap`: 겹치게 할 chunk의 길이
    - `length_function`: chunk의 길이를 측정할 함수
    - `is_seperator_regex`: `separators`를 정규식으로 볼 것인지 여부
- 작동 방식: 먼저 큰 단위의 구분자(`separators[0]`)로 텍스트를 나누고 각 분할된 chunk가 `chunk_size`를 초과하는 경우 다음 작은 단위의 구분자로 텍스트를 나눈다. 이 과정을 반복하여 모든 chunk가 `chunk_size` 이하가 될 때까지 분할한다.

 

 

문서 불러오기와 분할을 한 번에 수행하기

`loader.load_and_split`을 이용하여 불러오는 것과 동시에 분할을 수행한다.

loader = TextLoader('data/resell.txt', encoding='utf8')
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
split_docs = loader.load_and_split(text_splitter=text_splitter)

 

기존 방식과의 차이점이라면, `text_splitter.split_text(text)`을 하면 결과가 list로 나오는데 `loader.load_and_split(text_splitter=text_splitter)`를 하면 LangChain에 사용하는 Document로 나와 더 수월하다. LangChain에서 사용하려면 list에서 Document로 변환해줄 필요가 있다.

from langchain.docstore.document import Document

docs = [Document(page_content=text, metadata={"source": "local"}) for text in docs]

 

 

(3) 임베딩

OpenAI, AWS Bedrock(Titan), HuggingFace, Fast Embed 등 다양한 임베딩 모델이 있는데 본인의 필요에 따라 선택하여 사용하자. OpenAI와 Bedrock은 유료(OpenAI: ada v2 약 $0.1/1M 토큰, Bedrock: Titan Text Embeddings $0.0001/1M 토큰)이며 HuggingFace와 FastEmbed는 무료 모델이다.

 

 

임베딩 모델 비교

아래는 각 모델마다 문장을 임베딩 후 쿼리와 가장 가까운 문장을 추출하도록 실험한 내용이다.

임베딩 문장
1. 내가 구매하지 않은 엔비디아 주식이 오르다니, 배가 아프다.
2. 어젯밤에 뭘 잘못 먹었나... 배가 쓰리네.
3. 빙하랑 충돌해서 항해하던 배가 아야했어.
4. 하룻밤 사이에 비가 너무 쏟아져서 열려있던 배가 다쳤어. 그래서 팔 수가 없어.

일부러 헷갈리도록 "배가 아프다"는 표현을 다양하게 써봤다.

 

쿼리
A. 나도 매도 했어야했는데 이득 못 봐서 속상하네
B. 시장에서 산 밤을 까보니까 속이 비었네
C. 매운 걸 먹고 속이 너무 아파서 소화제를 먹었어
D. 사고난 자동차를 수리했어

내가 의도한 답은 A-1, B-4, C-2, D-3이다.

쿼리(질문)	A	B	C	D
HuggingFaceBge	1	4	4	3
FastEmbed	4	3	4	3
Bedrock(Titan)	4	4	2	3
OpenAI	3	4	2	3

결과는 Bedrock(Titan) = OpenAI > HuggingFaceBge > FastEmbed 이었다. 확실히 유료 모델의 성능이 더 좋은 것 같았으며 임베딩에 걸리는 시간이 중요하다면 OpenAI 모델을, 요금이 중요하다면 Bedrock을, 무료 모델을 쓴다면 HuggingFaceBge를 쓰는 것이 나을 것 같다는 생각이 들었다. 

 

임베딩 수행

LangChain을 통해 임베딩을 수행하는 코드는 간단하다.

from langchain_community.embeddings import HuggingFaceBgeEmbeddings, FastEmbedEmbeddings

hf = HuggingFaceBgeEmbeddings()
embeded_docs = hf.embed_documents([doc.page_content for doc in split_docs])

fe = FastEmbedEmbeddings()
embeded_docs = fe.embed_documents([doc.page_content for doc in split_docs])

문서를 HuggingFaceBge로 임베딩을 수행할 경우 1024 길이의 벡터로, FastEmbed로 임베딩을 수행할 경우 384 길이의 벡터로 변환된다.

 

# embeded_docs[0][:10]
[-0.02909066528081894,
 0.038625333458185196,
 0.039546430110931396,
 -0.026241781190037727,
 0.0429137758910656,
 -0.005134861450642347,
 -0.03184007853269577,
 -0.02016589231789112,
 0.005861369427293539,
 -0.029024090617895126]