코드로 알아보는 데이터 청킹!! : 5 Chunking Strategies for RAG with code

지난 시간에 이어서 직접 코드로 인공지능 검색 알고리즘을 위한 5가지 방법의 "데이터 청킹"을 해보겠습니다.

 

데이터 청킹이 어떤 것인지 감이 안 잡힌다면 다음 글 먼저 읽으시는 것을 권장 드립니다.

https://codzer.tistory.com/1

데이터 청킹(Ckunking)이란?

 

 

1. 검색 증강 모델(이하 RAG) 선언

from rich import print
from langchain.docstore.document import Document
from langchain_community.chat_models import ChatOllama
from langchain_community.vectorstores import Chroma
from langchain_community import embeddings
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

local_llm = ChatOllama(model="mistral") #로컬로 돌릴 수 있는 언어모델 가져옴

# RAG
def rag(chunks, collection_name):
    vectorstore = Chroma.from_documents(
        documents=documents,
        collection_name=collection_name,
        embedding=embeddings.ollama.OllamaEmbeddings(model='nomic-embed-text'),
    )
    retriever = vectorstore.as_retriever() #chroma(벡터 데이터베이스)에서 제공하는 retriever(검색기) 선언

    prompt_template = """Answer the question based only on the following context:
    {context}
    Question: {question}
    """
    prompt = ChatPromptTemplate.from_template(prompt_template)

    chain = (
        {"context": retriever, "question": RunnablePassthrough()} #retriever가 질문에 맞는 청크 찾아서 청크를 추가한 질문 전달
        | prompt #프롬프트 형식에 맞게 대입
        | local_llm #로컬 언어 모델이 처리
        | StrOutputParser()
    ) 
    result = chain.invoke("What is the use of Text Splitting?") #사용가가 질문을 하고 결과 받음
    print(result) #결과 출력

 

rag 함수를 보면 먼저 vactorstore를 선언하고 있습니다. chroma는 로컬 벡터데이터베이스로 document에 청크를 받고, embedding으로 각 청크들의 관계를 어떤 기준으로 정의할지 정합니다. 그리고 chroma에서 제공하는 함수로 retriever를 받을 수 있습니다. retriever는 검색기로 langchain에서 제공하는 라이브러리에 넣으면 해당 검색기로 데이터를 검색합니다.

 

프로세스(chain)를 보면

받은 입력과 관련된 청크를 받아와서 언어 모델 인공지능에게

"(찾은 자료) 이런 정보가 있는데 (질문)?"

이렇게 질문을 전달하고 있습니다.

그 결과를 받으면 RAG가 간단하게 완성 됩니다.

 

2. 청킹

그럼 이제부터 직접 청킹을 해보겠습니다! 청킹 결과는 글 가장 아래의 링크를 통해서 보실 수 있습니다.

2.1 Fixed-sized Chunking

# 1. Character Text Splitting
print("#### Character Text Splitting ####")

text = "Text splitting in LangChain is a critical feature that facilitates the division of large texts into smaller, manageable segments. "

# Manual Splitting
chunks = []
chunk_size = 35 # Characters
for i in range(0, len(text), chunk_size): #for문으로 일정 길이만큼 잘라서 chunks 리스트에 입력
    chunk = text[i:i + chunk_size]
    chunks.append(chunk)
documents = [Document(page_content=chunk, metadata={"source": "local"}) for chunk in chunks] #리스트를 langchain에서 제공하는 Document 형식으로
print(documents)

# Automatic Text Splitting
from langchain.text_splitter import CharacterTextSplitter #라이브러리를 이용한 방법
text_splitter = CharacterTextSplitter(chunk_size = 35, chunk_overlap=0, separator='', strip_whitespace=False)
documents = text_splitter.create_documents([text])
print(documents)

2.2 Recursive Chunking

# 2. Recursive Character Text Splitting
print("#### Recursive Character Text Splitting ####")

from langchain.text_splitter import RecursiveCharacterTextSplitter
with open('content.txt', 'r', encoding='utf-8') as file: # 대량의 텍스트로 이루어진 문서 호출
    text = file.read()

# ["\n\n", "\n", " ", ""]기준으로 청킹 실시, 라이브러리 이용
text_splitter = RecursiveCharacterTextSplitter(chunk_size = 65, chunk_overlap=0) 
print(text_splitter.create_documents([text]))

2.3 Documend Based Chunking

해당 문서 형식에 맞는 langChain 라이브러리를 불러옵니다. 문서 형식에 맞게 청킹된 결과가 나옵니다.

 

2.3.1 마크업(#나 \n 등)

# Document Specific Splitting - Markdown
from langchain.text_splitter import MarkdownTextSplitter
splitter = MarkdownTextSplitter(chunk_size = 40, chunk_overlap=0)
markdown_text = """
# Fun in California

## Driving

Try driving on the 1 down to San Diego

### Food

Make sure to eat a burrito while you're there

## Hiking

Go to Yosemite
"""
print(splitter.create_documents([markdown_text]))

 

2.3.2 파이썬 

# Document Specific Splitting - Python
from langchain.text_splitter import PythonCodeTextSplitter
python_text = """
class Person:
  def __init__(self, name, age):
    self.name = name
    self.age = age

p1 = Person("John", 36)

for i in range(10):
    print (i)
"""
python_splitter = PythonCodeTextSplitter(chunk_size=100, chunk_overlap=0)
print(python_splitter.create_documents([python_text]))

 

2.3.3 자바스크립트

# Document Specific Splitting - Javascript
from langchain.text_splitter import RecursiveCharacterTextSplitter, Language
javascript_text = """
// Function is called, the return value will end up in x
let x = myFunction(4, 3);

function myFunction(a, b) {
// Function returns the product of a and b
  return a * b;
}
"""
js_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.JS, chunk_size=65, chunk_overlap=0
)
print(js_splitter.create_documents([javascript_text]))

2.4 Sementic Chunking

# 4. Semantic Chunking
print("#### Semantic Chunking ####")

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings # 의미 관계 분석을 위한 기준

# Percentile - all differences between sentences are calculated, and then any difference greater than the X percentile is split
text_splitter = SemanticChunker(OpenAIEmbeddings())
text_splitter = SemanticChunker(
    OpenAIEmbeddings(), breakpoint_threshold_type="percentile" # "standard_deviation", "interquartile"
)
documents = text_splitter.create_documents([text])
print(documents)

 

OpenAI의 임베딩을 이용하여 청킹합니다. 의미 관계를 반영한 청킹이 가능합니다.

 

2.5 Agentic Chunking

저번 글에서 LLM을 활용한 청킹을 Agentic Chunking으로 정의했습니다. LLM을 활용한다고 해도 청크 구조는 사용자가 따로 지정해줘야합니다. 이번 시간에는 문장을 Proposition라는 것으로 변환하고, 이 Proposition을 카테고리 별로 묶어서 청킹하는 것을 LLM에게 시키겠습니다.

 

※Proposition

Proposition은 What Retrieval Granularity Should We Use?  라는 논문에서 제안한 효율적인 청킹 방식입니다. 간단하게 말하면, 한 문장으로 완벽한 맥락을 지니는 가장 작은 문장 단위로 나눠놓은 것을 Proposition이라고 하며 LLM에게 프롬프트를 입력하여 문서를 Proposition 단위로 바꿉니다. 이해가 잘 안 신다면 문장 별로 요약하는 방식이라고 생각하시면 됩니다. 이것 하나하나가 Proposition 입니다.

 

전반적인 처리 프로세스를 말해보자면

1. LLM에게 명령을 내려 Proposition 조건에 맞게 문장들을 Proposition으로 바꾼다.

2. LLM으로 각 Proposition를 카테고리화해서 묶는다.

3. 같은 카테고리의 문장들을 한 문단으로 합쳐서 저장한다.

 

이렇게 되면 같은 종류의 Proposition들은 같은 청크로 묶입니다.

 

먼저 문서를 Proposition으로 변환해줍니다.

# 5. Agentic Chunking
print("#### Proposition-Based Chunking ####")

from langchain.output_parsers.openai_tools import JsonOutputToolsParser
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain.chains import create_extraction_chain
from typing import Optional, List
from langchain.chains import create_extraction_chain_pydantic
from langchain_core.pydantic_v1 import BaseModel
from langchain import hub

obj = hub.pull("wfh/proposal-indexing") # proposition으로 바꿔주는 프롬프트 가져옴
llm = ChatOpenAI(model='gpt-4o-mini')
runnable = obj | llm

class Sentences(BaseModel):
    sentences: List[str]
    
# Extraction 우리가 원하는 형식(Sentences)으로 반환된 것만 가져옴 
extraction_chain = create_extraction_chain_pydantic(pydantic_schema=Sentences, llm=llm)
def get_propositions(text):
    runnable_output = runnable.invoke({
    	"input": text
    }).content
    propositions = extraction_chain.invoke(runnable_output)["text"][0].sentences
    return propositions
    
paragraphs = text.split("\n\n")
text_propositions = []
for i, para in enumerate(paragraphs[:5]):
    propositions = get_propositions(para)
    text_propositions.extend(propositions)
    print (f"Done with {i}")

print (f"You have {len(text_propositions)} propositions")
print(text_propositions[:10])

 

agentic_chunker는 다른 코드에서 불러옵니다. AgenticChunker의 내부 구조는

https://mer.vin/2024/03/chunking-strategy/

위 링크를 참조해주세요.

 

간단하게 기능만 설명하자면 받은 Proposition 들을 카테고리화 해서 묶어주는 역할을 합니다.

print("#### Agentic Chunking ####")

from agentic_chunker import AgenticChunker
ac = AgenticChunker()
# proposition을 리스트로 넣으면 카테고리화 후 같은 카테고리끼리 합쳐서 리스트로 반환
ac.add_propositions(text_propositions) 
print(ac.pretty_print_chunks())
chunks = ac.get_chunks(get_type='list_of_strings')
print(chunks)
documents = [Document(page_content=chunk, metadata={"source": "local"}) for chunk in chunks]
rag(documents, "agentic-chunks")

 

 

마무리

이렇게 해서 5가지의 청킹 방법을 직접 실습해보았습니다.

이해가 안 되는 부분이 있다면 얼마든지 댓글 달아주세요~!

 

참고

전체 코드와 실행 결과 : https://mer.vin/2024/03/chunking-strategy/

위 글은 다음 유튜브를 추가적으로 분석한 글입니다.

https://youtu.be/pIGRwMjhMaQ?si=NpaS5kCTIzfErPqF