[LangChain] 문서 로딩과 전처리 가이드

Document Loaders · Text Splitters · Cleaning & Normalization (LangChain)

RAG 파이프라인의 핵심은 “좋은 문서 입력”입니다. 어떤 소스에서 어떻게 로드하고, 어떻게 잘게 나눠(청킹) 저장하느냐가 검색 품질을 좌우합니다. 이 글에서는 LangChain에서 자주 쓰는 Document Loader와 Text Splitter, 그리고 Cleaning/Normalization 팁을 실전 코드와 함께 정리합니다.

본 글 예제는 LangChain 최신 구조를 기준으로 작성했습니다.

• 로더: langchain_community.document_loaders
• Google Drive: langchain_google_community
• 스플리터: langchain_text_splitters

---

0. 설치(필수/선택)

pip install -U langchain langchain-community langchain-text-splitters
# PDF
pip install pypdf
# 웹 파싱
pip install beautifulsoup4 lxml
# Confluence
pip install atlassian-python-api
# Google Drive (Google Workspace)
pip install langchain-google-community google-auth google-auth-oauthlib google-api-python-client
# (선택) 레이아웃 보존형 파서
pip install "unstructured[all-docs]"

---

1. Document Loaders — 소스별 로딩

LangChain의 로더는 문서를 Document(page_content, metadata) 리스트로 변환합니다. 이후 스플리터로 청킹하고 임베딩→벡터스토어에 저장하게 됩니다.

1-1. CSV (CSVLoader)

from langchain_community.document_loaders import CSVLoader

loader = CSVLoader(
    file_path="data/sample.csv",
    encoding="utf-8",
    csv_args={"delimiter": ","},   # 필요 시 구분자/quote 등 지정
    source_column=None                # 특정 컬럼을 source로 쓰고 싶으면 컬럼명 지정
)
docs = loader.load()
print(docs[0].page_content[:200], docs[0].metadata)

• 각 행(row)이 하나의 Document가 됩니다.
• source_column을 지정하면 해당 컬럼 값이 문서의 source 메타데이터로 들어가 추적성이 좋아집니다.

---

1-2. PDF (PyPDF / Unstructured)

가장 흔한 선택은 PyPDFLoader(가볍고 빠름). 레이아웃 보존이나 표, 헤더/푸터 처리가 중요하면 UnstructuredPDFLoader도 고려하세요.

from langchain_community.document_loaders import PyPDFLoader
# or: from langchain_community.document_loaders import UnstructuredPDFLoader

loader = PyPDFLoader("docs/whitepaper.pdf")
# loader = UnstructuredPDFLoader("docs/whitepaper.pdf", mode="elements")  # 레이아웃 보존형
docs = loader.load()
print(len(docs), docs[0].metadata)

• PyPDF는 텍스트 추출 중심, Unstructured는 레이아웃 요소를 elements로 유지 가능.
• 설치: pypdf (PyPDF) / unstructured[all-docs] (Unstructured).

---

1-3. 웹 문서 (WebBaseLoader)

from langchain_community.document_loaders import WebBaseLoader

loader = WebBaseLoader(
    web_paths=[
        "https://python.langchain.com/docs",
        "https://example.com/blog/post-1",
    ]
)
docs = loader.load()
print(docs[0].metadata.get("source"), docs[0].page_content[:200])

• HTML에서 본문 텍스트를 뽑아 page_content로 반환합니다.
• 사이트마다 구조가 달라 추가 후처리(선택자 필터/정규화)가 필요할 수 있습니다.

---

1-4. Confluence (ConfluenceLoader)

from langchain_community.document_loaders import ConfluenceLoader

loader = ConfluenceLoader(
    url="https://<your-domain>.atlassian.net/wiki",
    username="you@company.com",
    api_key="ATLASSIAN_API_TOKEN",      # Atlassian Personal Access Token
    cloud=True,                          # 서버(On-prem)는 cloud=False
    space_key="SPACE",                  # 혹은 page_ids=["12345", "67890"]
    limit=50,                            # 페이지 페이지네이션
    include_attachments=False,
)
docs = loader.load()
print(len(docs), docs[0].metadata)

• Cloud/Server 모두 지원, 스페이스/라벨/CQL 등 다양한 필터 옵션 제공.
• 대량 수집 시 limit, number_of_retries로 안정성을 조정하세요.

---

1-5. Google Workspace (Google Drive: Docs/Sheets/Slides/PDF)

Google Drive의 폴더/파일/문서 ID로 불러오며, 내부 파일 포맷은 보통 UnstructuredFileIOLoader로 파싱합니다. 권장 임포트는 langchain_google_community 입니다.

from langchain_google_community import GoogleDriveLoader
from langchain_community.document_loaders import UnstructuredFileIOLoader

# 단일 파일
loader = GoogleDriveLoader(
    file_ids=["<google-file-id>"],
    file_loader_cls=UnstructuredFileIOLoader,
    file_loader_kwargs={"mode": "elements"}  # 표/헤더 등 레이아웃 요소 보존
)
docs = loader.load()

# 폴더 전체(하위 포함)
folder_loader = GoogleDriveLoader(
    folder_id="<google-folder-id>",
    recursive=True,
    file_loader_cls=UnstructuredFileIOLoader,
    file_loader_kwargs={"mode": "elements"},
)
folder_docs = folder_loader.load()

• Docs/Sheets/Slides/PDF 혼재 폴더도 처리 가능(파일 유형별로 파서 적용).
• 인증은 서비스 계정 또는 OAuth를 사용합니다(Drive API 활성화 필요).
• langchain_community.document_loaders.googledrive.GoogleDriveLoader는 deprecated 안내가 있어 langchain_google_community.GoogleDriveLoader 사용을 권장합니다.

---

2. Text Splitters — 청킹 전략

로드한 문서는 길이가 제각각입니다. 모델/검색기에 맞게 일정한 크기로 나눠야 하고, 중첩(overlap) 으로 문맥을 잇는 게 중요합니다.

2-1. 기본기: RecursiveCharacterTextSplitter

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,           # 문서 길이/도메인에 맞게 조정
    chunk_overlap=120,        # 문맥 유지를 위한 겹침
    separators=["\n\n", "\n", " ", ""],  # 문단→문장→단어 순서로 시도
)
chunks = splitter.split_documents(docs)  # docs: 위에서 로드한 Document 리스트
print(len(chunks), chunks[0].page_content[:200])

• 문단/문장 경계를 최대한 유지하며 크기 조건을 만족할 때까지 재귀적으로 분할.
• 대부분의 일반 텍스트에 첫 손에 꼽히는 선택입니다.

2-2. 토큰 기준 분할: TokenTextSplitter (+ tiktoken 계열)

from langchain_text_splitters import TokenTextSplitter

token_splitter = TokenTextSplitter(chunk_size=400, chunk_overlap=40)
token_chunks = token_splitter.split_documents(docs)

• 모델 입력 한도를 토큰 기준으로 딱 맞추고 싶을 때 사용.
• 일부 언어(한/중/일)에서는 토큰 단위 split 시 유니코드 이슈가 있을 수 있어 from_tiktoken_encoder 계열 사용을 권장합니다.

2-3. 언어/도메인 특화 분할(선택)

• NLTK/KoNLPy/Spacy/SentenceTransformers 토크나이저 기반 분할 클래스도 제공됩니다. 대화/문장 경계가 중요한 데이터에 유용합니다.

팁

• 청크 크기는 임베딩 모델/질의 유형에 따라 다릅니다. 규칙 하나로 고정하기보다, 오프라인 리콜·정확도 테스트(LangSmith 평가 등)로 데이터 기반 튜닝을 추천합니다.

---

3. Cleaning & Normalization — 질 좋은 텍스트 만들기

웹/문서 원본은 노이즈가 많습니다. 아래 전처리로 임베딩 품질과 검색 정밀도를 끌어올릴 수 있습니다.

3-1. HTML/공백/제어문자 정리

import re
from bs4 import BeautifulSoup
from langchain_core.documents import Document

ZWS = u"\u200b"  # zero-width space
ZWNJ = u"\u200c"
ZWJ = u"\u200d"

def clean_text(t: str) -> str:
    # 1) HTML 제거
    if "<html" in t.lower() or "</" in t:
        t = BeautifulSoup(t, "lxml").get_text(separator="\n")
    # 2) 제어문자/제로폭 문자 제거
    t = t.replace(ZWS, "").replace(ZWNJ, "").replace(ZWJ, "")
    t = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F]", "", t)
    # 3) 공백 정규화 (연속 공백/개행 축소)
    t = re.sub(r"[ \t]+", " ", t)
    t = re.sub(r"\n{3,}", "\n\n", t).strip()
    return t

def normalize_docs(docs):
    out = []
    for d in docs:
        text = clean_text(d.page_content)
        # (선택) 불릿統一: •, ·, - → "-"
        text = re.sub(r"[•·‧∙●◦▪▫➤►▶]", "-", text)
        out.append(Document(page_content=text, metadata=d.metadata))
    return out

# 로드 직후 한 번 통과
docs = normalize_docs(docs)

3-2. 중복/길이·언어 필터

def dedupe_by_content(docs):
    seen, out = set(), []
    for d in docs:
        key = hash(d.page_content)
        if key not in seen:
            seen.add(key)
            out.append(d)
    return out

def filter_by_length(docs, min_chars=50):
    return [d for d in docs if len(d.page_content) >= min_chars]

docs = dedupe_by_content(docs)
docs = filter_by_length(docs, min_chars=80)

3-3. 메타데이터 보강

def add_metadata_defaults(docs, source_default="unknown", doc_type="web"):
    for d in docs:
        d.metadata.setdefault("source", source_default)
        d.metadata.setdefault("doc_type", doc_type)
    return docs

docs = add_metadata_defaults(docs, source_default="confluence", doc_type="kb")

팁

• PDF→텍스트 추출 품질이 낮으면 Unstructured로 바꾸거나, 표/수식 많은 문서는 mode="elements"로 구조를 보존해보세요.
• 웹 크롤링 후에는 중복(내비/푸터/사이드바 텍스트) 제거가 큰 효과를 냅니다.

---

4. End-to-End 예시 (다중 소스 통합 → 전처리 → 청킹)

from langchain_community.document_loaders import CSVLoader, PyPDFLoader, WebBaseLoader
from langchain_community.document_loaders import ConfluenceLoader, UnstructuredFileIOLoader
from langchain_google_community import GoogleDriveLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 1) 다양한 소스에서 로드
csv_docs = CSVLoader("data/faqs.csv").load()
pdf_docs = PyPDFLoader("docs/guide.pdf").load()
web_docs = WebBaseLoader(["https://example.com/blog"]).load()
conf_docs = ConfluenceLoader(
    url="https://your.atlassian.net/wiki",
    username="you@company.com",
    api_key="ATLASSIAN_API_TOKEN",
    space_key="SPACE",
).load()
gdocs = GoogleDriveLoader(
    folder_id="",
    recursive=True,
    file_loader_cls=UnstructuredFileIOLoader,
    file_loader_kwargs={"mode": "elements"},
).load()

docs = csv_docs + pdf_docs + web_docs + conf_docs + gdocs

# 2) 클리닝 & 정규화
docs = normalize_docs(docs)
docs = dedupe_by_content(docs)
docs = filter_by_length(docs, min_chars=80)

# 3) 청킹
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=120)
chunks = splitter.split_documents(docs)

print(f"총 원문 {len(docs)} → 청크 {len(chunks)}")
print(chunks[0].metadata)
print(chunks[0].page_content[:300])

---

마무리

• 로더는 소스 특성을 이해하고(예: Confluence 라벨/CQL, Google Drive 폴더/파일 혼재) 올바른 파서를 고르는 게 핵심.
• 스플리터는 도메인·모델 제약에 맞게 크기/중첩을 데이터 기반으로 튜닝.
• 클리닝/정규화는 작은 정성으로 큰 성능 향상을 줍니다(중복 제거, 공백/HTML/표준화).

사내 Confluence/Google Workspace 인증 스니펫(토큰·스코프·보안 가이드 포함)도 필요하시면 맞춤으로 추가해 드릴게요.