RecursiveCharacterTextSplitter - chunk_size, chunk_overlap 설정 (251017)

현재 build_vector_store 함수는 단순히 1000자씩 잘라내는 슬라이싱 방식으로 텍스트를 분할하고 있는데요 — 이 방식은 문장이나 문단이 중간에 끊기는 문제가 있어, 검색 품질이 떨어질 수 있습니다.

이를 개선하기 위해 LangChain의

from langchain.text_splitter import RecursiveCharacterTextSplitter

를 사용하는 것이 훨씬 좋은 선택입니다.

아래는 ✅ Chunk size / overlap을 인자로 조정할 수 있게 개선한 코드입니다.
또한, RecursiveCharacterTextSplitter를 사용하지 않고도 하는 방법도 아래에 함께 설명드리겠습니다.

✅ 개선된 코드 (RecursiveCharacterTextSplitter 적용)

import os
import glob
import PyPDF2
from tqdm import tqdm
import chromadb
from sentence_transformers import SentenceTransformer
from langchain.text_splitter import RecursiveCharacterTextSplitter
from google import genai


def extract_text_from_pdf(pdf_path):
    text = ""
    with open(pdf_path, "rb") as file:
        reader = PyPDF2.PdfReader(file)
        for page in reader.pages:
            page_text = page.extract_text()
            if page_text:
                text += page_text + "\n"
    return text


def build_vector_store(
    doc_folder="./doc",
    db_path="./chroma_db",
    chunk_size=1000,
    chunk_overlap=100,
):
    """
    PDF 폴더를 읽어 ChromaDB에 임베딩 저장.
    - chunk_size: 청크 크기 (문자 수 단위)
    - chunk_overlap: 청크 간 겹치는 영역 크기
    """
    print(f"📚 Building ChromaDB persistent store from: {doc_folder}")

    client = chromadb.PersistentClient(path=db_path)
    collection = client.get_or_create_collection(name="pdf_docs")

    embedder = SentenceTransformer("all-MiniLM-L6-v2")

    pdf_files = glob.glob(os.path.join(doc_folder, "*.pdf"))
    if not pdf_files:
        print("⚠️ No PDF files found in ./doc folder.")
        return

    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", ".", " ", ""],  # 문단 > 문장 > 공백 > 글자 단위로 점진 분할
    )

    for pdf_path in tqdm(pdf_files, desc="Processing PDFs"):
        text = extract_text_from_pdf(pdf_path)
        chunks = text_splitter.split_text(text)

        embeddings = embedder.encode(chunks, convert_to_numpy=True).tolist()
        ids = [f"{os.path.basename(pdf_path)}_{i}" for i in range(len(chunks))]
        collection.add(ids=ids, documents=chunks, embeddings=embeddings)

    print("✅ Vector store built successfully and saved persistently.")


def chat_with_agent(query, db_path="./chroma_db"):
    print(f"💬 Query: {query}")

    client = chromadb.PersistentClient(path=db_path)
    collection = client.get_or_create_collection(name="pdf_docs")

    embedder = SentenceTransformer("all-MiniLM-L6-v2")
    query_emb = embedder.encode(query).tolist()

    results = collection.query(query_embeddings=[query_emb], n_results=3)
    docs = results.get("documents", [[]])[0]
    if not docs:
        print("⚠️ No relevant documents found.")
        return

    context = "\n".join(docs)

    try:
        genai_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
        response = genai_client.models.generate_content(
            model="gemini-2.0-flash",
            contents=f"다음 문서를 참고하여 질문에 답하세요:\n{context}\n\n질문: {query}",
        )
        print("🧠 Agent Response:\n", response.text)
    except Exception as e:
        print("⚠️ Google GenAI error:", e)
        print("🔍 Fallback Answer:")
        print(context[:1000])


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument("--build", action="store_true", help="Build ChromaDB index from PDFs")
    parser.add_argument("--chat", type=str, help="Ask a question to the RAG agent", default=None)
    parser.add_argument("--chunk_size", type=int, default=1000, help="Chunk size for text splitting")
    parser.add_argument("--chunk_overlap", type=int, default=100, help="Chunk overlap for text splitting")
    args = parser.parse_args()

    if args.build:
        build_vector_store(chunk_size=args.chunk_size, chunk_overlap=args.chunk_overlap)
    elif args.chat:
        chat_with_agent(args.chat)
    else:
        print("⚙️ Usage: python rag_agent_chroma_adk.py --build [--chunk_size 1000 --chunk_overlap 100] | --chat '질문'")

✅ 실행 예시

# 기본 설정으로 벡터스토어 구축
python rag_agent_chroma_adk.py --build

# 청크 크기/오버랩 조정
python rag_agent_chroma_adk.py --build --chunk_size 800 --chunk_overlap 150

이렇게 하면 문단 단위로 깔끔하게 분할되며, 문장이 자연스럽게 이어져서 RAG 검색 품질이 올라갑니다.

🚀 대안: RecursiveCharacterTextSplitter 없이 구현하는 방법

RecursiveCharacterTextSplitter를 쓰지 않고도 다음과 같은 방식으로 커스텀 분할이 가능합니다.

def simple_text_split(text, chunk_size=1000, overlap=100):
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start += chunk_size - overlap  # 다음 chunk는 overlap만큼 겹치게 이동
    return chunks

이 함수는 훨씬 단순하지만, 문장 단위로 끊지 않기 때문에 의미가 잘리는 단점이 있습니다.
그래서 가장 권장되는 방법은 RecursiveCharacterTextSplitter를 사용하는 것입니다.

추가 내용 (계속)

📘 “PDF에서 문단 기준으로만 청크 나누기”
📘 “한글 텍스트 분리 최적화 버전” (예: KSS 라이브러리 활용)

영문 기반의 RecursiveCharacterTextSplitter는 문장 구분을 영어 문장부호(., !, ?) 중심으로 하기 때문에, 한국어 PDF 텍스트를 처리할 때는 문장이 이상하게 끊기거나, 조사·어미가 잘리는 경우가 자주 발생합니다.

그래서 한국어 문서를 처리할 때는 한국어 문장 단위로 정확히 분리할 수 있는 도구, 예를 들어 KSS (Korean Sentence Splitter)를 사용하는 것이 훨씬 낫습니다.

✅ KSS를 이용한 “한글 텍스트 분리 최적화 버전”

아래는 기존의 RecursiveCharacterTextSplitter 대신 kss를 사용하여 문장 단위로 잘게 나눈 다음, chunk size와 overlap을 조정해주는 코드입니다.

import os
import glob
import PyPDF2
from tqdm import tqdm
import chromadb
import kss
from sentence_transformers import SentenceTransformer
from google import genai


def extract_text_from_pdf(pdf_path):
    text = ""
    with open(pdf_path, "rb") as file:
        reader = PyPDF2.PdfReader(file)
        for page in reader.pages:
            page_text = page.extract_text()
            if page_text:
                text += page_text + "\n"
    return text


def split_korean_text(text, chunk_size=1000, chunk_overlap=100):
    """
    한국어 문장 단위로 나눈 후 chunk_size 기준으로 묶어주는 함수
    """
    sentences = list(kss.split_sentences(text))
    chunks = []
    current_chunk = ""

    for sent in sentences:
        if len(current_chunk) + len(sent) < chunk_size:
            current_chunk += sent + " "
        else:
            chunks.append(current_chunk.strip())
            # overlap 적용 (마지막 부분 일부 유지)
            if chunk_overlap > 0 and chunks:
                overlap_text = current_chunk[-chunk_overlap:]
                current_chunk = overlap_text + sent + " "
            else:
                current_chunk = sent + " "
    if current_chunk:
        chunks.append(current_chunk.strip())

    return chunks


def build_vector_store(
    doc_folder="./doc",
    db_path="./chroma_db",
    chunk_size=1000,
    chunk_overlap=100,
):
    print(f"📚 Building ChromaDB persistent store from: {doc_folder}")

    client = chromadb.PersistentClient(path=db_path)
    collection = client.get_or_create_collection(name="pdf_docs")

    embedder = SentenceTransformer("all-MiniLM-L6-v2")

    pdf_files = glob.glob(os.path.join(doc_folder, "*.pdf"))
    if not pdf_files:
        print("⚠️ No PDF files found in ./doc folder.")
        return

    for pdf_path in tqdm(pdf_files, desc="Processing PDFs"):
        text = extract_text_from_pdf(pdf_path)
        chunks = split_korean_text(text, chunk_size, chunk_overlap)

        embeddings = embedder.encode(chunks, convert_to_numpy=True).tolist()
        ids = [f"{os.path.basename(pdf_path)}_{i}" for i in range(len(chunks))]
        collection.add(ids=ids, documents=chunks, embeddings=embeddings)

    print("✅ Vector store built successfully with KSS-based Korean sentence splitting.")


def chat_with_agent(query, db_path="./chroma_db"):
    print(f"💬 Query: {query}")

    client = chromadb.PersistentClient(path=db_path)
    collection = client.get_or_create_collection(name="pdf_docs")

    embedder = SentenceTransformer("all-MiniLM-L6-v2")
    query_emb = embedder.encode(query).tolist()

    results = collection.query(query_embeddings=[query_emb], n_results=3)
    docs = results.get("documents", [[]])[0]
    if not docs:
        print("⚠️ No relevant documents found.")
        return

    context = "\n".join(docs)

    try:
        genai_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
        response = genai_client.models.generate_content(
            model="gemini-2.0-flash",
            contents=f"다음 문서를 참고하여 질문에 답하세요:\n{context}\n\n질문: {query}",
        )
        print("🧠 Agent Response:\n", response.text)
    except Exception as e:
        print("⚠️ Google GenAI error:", e)
        print("🔍 Fallback Answer:")
        print(context[:1000])


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument("--build", action="store_true", help="Build ChromaDB index from PDFs")
    parser.add_argument("--chat", type=str, help="Ask a question to the RAG agent", default=None)
    parser.add_argument("--chunk_size", type=int, default=1000, help="Chunk size for text splitting")
    parser.add_argument("--chunk_overlap", type=int, default=100, help="Chunk overlap for text splitting")
    args = parser.parse_args()

    if args.build:
        build_vector_store(chunk_size=args.chunk_size, chunk_overlap=args.chunk_overlap)
    elif args.chat:
        chat_with_agent(args.chat)
    else:
        print("⚙️ Usage: python rag_agent_chroma_adk.py --build [--chunk_size 1000 --chunk_overlap 100] | --chat '질문'")

✅ KSS 설치 방법

pip install kss

💡 kss는 GPU 없이도 빠르게 동작하며, 한국어 맞춤형 문장 분리 모델을 사용합니다.

🔍 이 방식의 장점

🚀 추천 조합

추가

KSS 버전을 RecursiveCharacterTextSplitter와 선택적으로 토글 가능한 하이브리드 버전으로 작성할 수 있다.

예:

python rag_agent_chroma_adk.py --build --splitter kss
python rag_agent_chroma_adk.py --build --splitter recursive