PDF에서 문단(Paragraph) 단위로만 청크 나누기 (251017)

PDF에서 문단(Paragraph) 단위로만 청크를 나누는 버전

이 방식은 KSS처럼 문장을 세세하게 쪼개지 않고, 빈 줄(줄바꿈 2회 이상) 또는 긴 줄바꿈 구간을 기준으로 자연스러운 문단 단위 분리를 수행합니다. 즉, 보고서·논문·에세이처럼 문단 구조가 뚜렷한 PDF에 특히 잘 맞습니다.

✅ 코드: PDF 문단 기준 청크 분리 버전

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


def extract_text_from_pdf(pdf_path):
    """PDF의 모든 페이지를 읽어 텍스트로 변환"""
    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_text_by_paragraph(text, chunk_size=1000, chunk_overlap=100):
    """
    문단(빈 줄 2개 이상) 기준으로 분리한 뒤,
    너무 긴 문단은 chunk_size 기준으로 나누고, chunk_overlap 적용
    """
    # 1️⃣ 문단 단위로 나누기 (빈 줄 2개 이상 또는 \n\n 기준)
    paragraphs = re.split(r"\n\s*\n+", text.strip())

    chunks = []
    for para in paragraphs:
        para = para.strip()
        if not para:
            continue

        # 2️⃣ 너무 긴 문단은 chunk_size 기준으로 추가 분할
        if len(para) > chunk_size:
            start = 0
            while start < len(para):
                end = start + chunk_size
                chunk = para[start:end]
                chunks.append(chunk)
                start += chunk_size - chunk_overlap
        else:
            chunks.append(para)

    return chunks


def build_vector_store(
    doc_folder="./doc",
    db_path="./chroma_db",
    chunk_size=1000,
    chunk_overlap=100,
):
    """
    PDF 폴더 내 문서들을 문단 기준으로 분할 후 ChromaDB에 임베딩 저장
    """
    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_text_by_paragraph(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 paragraph-based 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 (paragraph-based)")
    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 splitting long paragraphs")
    parser.add_argument("--chunk_overlap", type=int, default=100, help="Overlap between chunks for long paragraphs")
    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

# 긴 문단을 1500자 기준으로 분할
python rag_agent_chroma_adk.py --build --chunk_size 1500 --chunk_overlap 200

🔍 장점 / 단점 비교

💡 추천 조합

• 보고서·연구 논문·정책 문서 → 문단 기준 버전
• 뉴스·인터뷰·대화체 → KSS 문장 기준 버전
• 영문/혼합 문서 → RecursiveCharacterTextSplitter

세 가지 분리 방식을 하나의 통합 스크립트에서 옵션으로 선택할 수 있게 만들어드릴 수도 있습니다.
예:

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