데이터가 전부입니다 - 문서를 지식으로 바꾸는 파이프라인

소규모 기업을 위한 제대로 된 RAG 시스템 [2/4]

1편에서 기본 RAG가 왜 실패하는지, 그리고 프로덕션에서 검증된 아키텍처가 어떤 모습인지 살펴봤습니다. 결론은 명확했습니다. RAG 실패의 80%는 검색이나 생성 단계가 아니라, 데이터를 준비하는 단계에서 비롯됩니다.

"쓰레기를 넣으면 쓰레기가 나온다." 프로그래밍을 시작한 날부터 들어온 격언인데, 이것이 RAG에서는 유독 잔인하게 적용됩니다. LLM은 검색 시스템이 건네준 컨텍스트를 "이건 시스템이 가져온 거니까 맞겠지"라고 믿고 답변을 만들어냅니다. 검색 시스템이 엉뚱한 청크를 가져왔어도 LLM은 그걸 기반으로 자신있게 답합니다. 문맥이 잘린 청크가 들어오면 LLM이 빠진 문맥을 제멋대로 채워넣습니다. 원본 문서의 테이블이 깨져서 숫자가 뒤섞여 들어오면 LLM은 그 뒤섞인 숫자를 근거로 분석합니다.

Stratagem Systems의 RAG 비용 분석 보고서에 따르면, 데이터 정리와 전처리가 RAG 프로젝트 비용의 30~50%를 차지합니다. 처음 이 수치를 보면 놀랍지만, 실제로 RAG 시스템을 구축해보면 오히려 보수적인 추정이라고 느끼는 개발자가 많습니다. 개발자 커뮤니티에서 "RAG 프로젝트의 60%는 RAG가 필요한 것이 아니라 더 나은 검색이 필요하다"는 말이 도는 이유도 같습니다. 그리고 더 나은 검색의 시작은 더 나은 데이터 준비입니다.

이번 편에서는 데이터 수집 계층을 다룹니다. 문서를 수집하고, 전처리하고, 청킹하고, 문맥을 붙이고, 임베딩으로 변환해서 인덱싱하는 전체 과정입니다. 소규모 기업이 실제로 돌릴 수 있는 코드를 함께 작성합니다.

다음 그림은 이번 편에서 구축할 데이터 수집 파이프라인의 전체 흐름입니다. 6단계로 구성되며, Ollama 로컬 실행으로 API 비용 없이 운영할 수 있습니다.

문서 로딩부터 벡터 인덱싱까지 각 단계를 순서대로 살펴보겠습니다. 전처리와 청킹이 전체 품질을 좌우하는 핵심 단계이고, Contextual Retrieval로 문맥을 보강한 뒤 임베딩과 인덱싱으로 마무리합니다.

문서 전처리 - 보이지 않는 80%의 노력

RAG 튜토리얼의 첫 줄은 대부분 이렇게 시작합니다.

복사
from langchain.document_loaders import TextLoader
loader = TextLoader("document.txt")
docs = loader.load()

현실의 사내 문서는 .txt가 아닙니다. PDF, 워드, 엑셀, 파워포인트, 노션 페이지, 구글 독스, 위키, HTML, 마크다운이 섞여 있습니다. 이 형식들을 일관된 텍스트로 변환하는 게 첫 번째 관문이고, 여기서부터 품질 격차가 벌어집니다.

PDF가 가장 까다롭습니다. PDF는 겉보기엔 문서지만 내부적으로는 페이지 위에 텍스트, 이미지, 도형을 좌표로 배치한 구조입니다. 2단 컬럼 레이아웃의 PDF를 단순 텍스트 추출하면 왼쪽 컬럼의 첫 줄과 오른쪽 컬럼의 첫 줄이 한 줄로 합쳐집니다. 테이블의 셀 경계가 무시되어 숫자가 뒤섞입니다. 헤더와 푸터가 본문 사이에 끼어듭니다.

LlamaIndex의 LlamaParse는 이 문제를 전용 모델로 해결합니다. 테이블 구조, 다중 컬럼 레이아웃, 기울어진 스캔 문서까지 구조를 보존하면서 텍스트를 추출합니다. 2025년에 skew(기울기) 감지와 신규 모델이 추가되었습니다. 무료 티어에서 하루 1,000페이지까지 처리할 수 있으므로 소규모 기업에 현실적인 선택입니다.

다음은 기본적인 문서 로딩 파이프라인 코드입니다. 여러 형식의 문서를 하나의 파이프라인으로 처리합니다.

복사
# document_loader.py
import os
from pathlib import Path
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class Document:
    """통합 문서 표현."""
    content: str
    metadata: dict = field(default_factory=dict)
    source: str = ""
    doc_type: str = ""

class DocumentLoader:
    """여러 형식의 문서를 통합 로딩하는 클래스."""

    def __init__(self, use_llama_parse: bool = False, llama_parse_api_key: str = ""):
        self.use_llama_parse = use_llama_parse
        self.llama_parse_api_key = llama_parse_api_key

    def load_directory(self, dir_path: str) -> list[Document]:
        """디렉토리 내 모든 지원 형식의 문서를 로딩합니다."""
        documents = []
        supported_extensions = {".pdf", ".docx", ".md", ".txt", ".html", ".csv"}

        for file_path in Path(dir_path).rglob("*"):
            if file_path.suffix.lower() in supported_extensions:
                try:
                    doc = self.load_file(str(file_path))
                    if doc:
                        documents.append(doc)
                except Exception as e:
                    print(f"[WARN] {file_path} 로딩 실패: {e}")

        print(f"총 {len(documents)}개 문서 로딩 완료")
        return documents

    def load_file(self, file_path: str) -> Optional[Document]:
        """단일 파일을 로딩합니다."""
        ext = Path(file_path).suffix.lower()
        loader_map = {
            ".pdf": self._load_pdf,
            ".docx": self._load_docx,
            ".md": self._load_text,
            ".txt": self._load_text,
            ".html": self._load_html,
            ".csv": self._load_csv,
        }

        loader = loader_map.get(ext)
        if not loader:
            return None

        content = loader(file_path)
        if not content or not content.strip():
            return None

        return Document(
            content=content,
            metadata={
                "file_name": Path(file_path).name,
                "file_path": file_path,
                "file_type": ext,
                "file_size": os.path.getsize(file_path),
            },
            source=file_path,
            doc_type=ext,
        )

    def _load_pdf(self, file_path: str) -> str:
        if self.use_llama_parse:
            return self._load_pdf_with_llama_parse(file_path)
        return self._load_pdf_with_pymupdf(file_path)

    def _load_pdf_with_pymupdf(self, file_path: str) -> str:
        """PyMuPDF로 PDF 텍스트 추출. 단순 PDF에 적합합니다."""
        import fitz  # pymupdf
        doc = fitz.open(file_path)
        text_parts = []
        for page_num, page in enumerate(doc):
            text = page.get_text("text")
            if text.strip():
                text_parts.append(f"[페이지 {page_num + 1}]\n{text}")
        doc.close()
        return "\n\n".join(text_parts)

    def _load_pdf_with_llama_parse(self, file_path: str) -> str:
        """LlamaParse로 PDF 구조 보존 추출. 테이블, 다중 컬럼에 적합합니다."""
        from llama_parse import LlamaParse
        parser = LlamaParse(
            api_key=self.llama_parse_api_key,
            result_type="markdown",
            verbose=False,
        )
        documents = parser.load_data(file_path)
        return "\n\n".join(doc.text for doc in documents)

    def _load_docx(self, file_path: str) -> str:
        from docx import Document as DocxDocument
        doc = DocxDocument(file_path)
        return "\n\n".join(para.text for para in doc.paragraphs if para.text.strip())

    def _load_text(self, file_path: str) -> str:
        with open(file_path, "r", encoding="utf-8") as f:
            return f.read()

    def _load_html(self, file_path: str) -> str:
        from bs4 import BeautifulSoup
        with open(file_path, "r", encoding="utf-8") as f:
            soup = BeautifulSoup(f.read(), "html.parser")
        # 스크립트, 스타일 태그 제거
        for tag in soup(["script", "style", "nav", "footer", "header"]):
            tag.decompose()
        return soup.get_text(separator="\n", strip=True)

    def _load_csv(self, file_path: str) -> str:
        import csv
        with open(file_path, "r", encoding="utf-8") as f:
            reader = csv.reader(f)
            rows = list(reader)
        if not rows:
            return ""
        # 첫 행을 헤더로 처리
        headers = rows[0]
        text_parts = []
        for row in rows[1:]:
            pairs = [f"{h}: {v}" for h, v in zip(headers, row) if v.strip()]
            text_parts.append(", ".join(pairs))
        return "\n".join(text_parts)

이 코드에서 주목할 점은 PDF 처리 분기입니다. 단순한 텍스트 위주 PDF는 PyMuPDF로 충분하지만, 테이블이나 복잡한 레이아웃이 있는 PDF는 LlamaParse를 사용합니다. NVIDIA의 2024년 벤치마크에서 페이지 단위 PDF 처리가 0.648 정확도로 최고 점수를 기록한 것도, PDF의 구조를 제대로 보존했기 때문입니다.

전처리에서 간과하기 쉬운 부분이 메타데이터입니다. 문서의 내용만 추출하는 것이 아니라, 문서가 어디서 왔는지, 언제 만들어졌는지, 어떤 부서의 것인지, 어떤 카테고리인지를 함께 저장해야 합니다. 이 메타데이터가 나중에 검색 필터로 쓰입니다. "인사팀 문서에서만 검색", "2025년 이후 문서만 검색" 같은 조건을 메타데이터 없이는 구현할 수 없습니다.

복사
# metadata_enricher.py
from datetime import datetime
from pathlib import Path

class MetadataEnricher:
    """문서에 구조화된 메타데이터를 추가합니다."""

    def __init__(self, department_mapping: dict = None):
        # 디렉토리 경로와 부서를 매핑
        self.department_mapping = department_mapping or {}

    def enrich(self, doc: Document) -> Document:
        """문서에 메타데이터를 추가합니다."""
        file_path = Path(doc.source)

        # 기본 메타데이터
        doc.metadata["created_at"] = datetime.fromtimestamp(
            file_path.stat().st_ctime
        ).isoformat()
        doc.metadata["modified_at"] = datetime.fromtimestamp(
            file_path.stat().st_mtime
        ).isoformat()

        # 디렉토리 기반 부서 매핑
        for dir_pattern, department in self.department_mapping.items():
            if dir_pattern in str(file_path):
                doc.metadata["department"] = department
                break

        # 문서 통계
        doc.metadata["char_count"] = len(doc.content)
        doc.metadata["word_count"] = len(doc.content.split())
        doc.metadata["line_count"] = doc.content.count("\n") + 1

        return doc

Royal Bank of Canada의 Arcane 시스템 사례에서 배울 점이 있습니다. 이 시스템의 가장 큰 기술적 도전은 "웹 플랫폼, 독점 소스, PDF, Excel에 분산된 데이터를 통합하는 것"이었습니다. 소규모 기업도 사정은 같습니다. 구글 드라이브, 노션, 사내 위키, 공유 폴더에 문서가 흩어져 있고, 같은 내용이 여러 버전으로 존재하기도 합니다. 전처리 단계에서 중복 문서를 감지하고, 최신 버전만 남기는 로직이 필요합니다.

중복 감지는 간단한 해시 비교만으로도 상당 부분 해결됩니다.

복사
import hashlib

class DuplicateDetector:
    """문서 중복을 감지합니다."""

    def __init__(self):
        self.seen_hashes: dict[str, str] = {}

    def is_duplicate(self, doc: Document) -> bool:
        """문서가 이미 처리된 것과 중복인지 확인합니다."""
        content_hash = hashlib.sha256(doc.content.encode()).hexdigest()

        if content_hash in self.seen_hashes:
            original = self.seen_hashes[content_hash]
            print(f"[중복] {doc.source} == {original}")
            return True

        self.seen_hashes[content_hash] = doc.source
        return False

    def filter_duplicates(self, documents: list[Document]) -> list[Document]:
        """중복을 제거한 문서 목록을 반환합니다."""
        unique = []
        for doc in documents:
            if not self.is_duplicate(doc):
                unique.append(doc)
        print(f"중복 제거: {len(documents)} -> {len(unique)}개")
        return unique

이 구현은 내용이 완전히 동일한 문서만 감지합니다. 현실에서 더 빈번한 문제는 near-duplicate, 즉 파일명만 다르거나 날짜만 업데이트된 거의 같은 문서입니다. 이런 유사 문서까지 감지하려면 SimHash나 MinHash 같은 근사 중복 감지 알고리즘을 도입해야 합니다. 다만 소규모 기업에서 처음 시작할 때는 해시 비교만으로도 상당 부분 해결되므로, 검색 결과에 중복 내용이 눈에 띄게 나타나기 시작하면 그때 도입을 검토합니다.

전처리는 화려하지 않습니다. 눈에 보이는 기능이 아니라서 시간 투자를 꺼리는 팀이 많은데, 여기서 품질이 무너지면 그 뒤의 모든 단계가 소용없습니다. RAG 프로젝트를 진행해 본 개발자에게 물어보면 거의 다 같은 이야기를 합니다. "전처리에 시간을 더 쏟았어야 했다."

청킹, RAG의 생사를 결정하는 선택

전처리가 끝나면 문서를 검색 가능한 단위로 나누어야 합니다. 이 과정을 청킹이라고 부르며, RAG 시스템 전체에서 가장 결정적인 단계입니다.

왜 청킹이 이렇게 중요한 겁니까? LLM의 컨텍스트 윈도우에는 한계가 있습니다. 문서 전체를 넣을 수 없으니 관련된 부분만 잘라서 넣어야 합니다. 이때 "어떻게 자르느냐"가 검색 품질을 직접 결정합니다. 너무 작게 자르면 문맥이 사라지고, 너무 크게 자르면 관련 없는 내용이 함께 딸려옵니다. 1편에서 언급한 대로 RAG 실패의 80%가 청킹 결정에서 비롯된다는 분석이 있을 정도입니다.

2026년 기준으로 실무에서 검토할 만한 청킹 전략을 하나씩 살펴보겠습니다.

다음 그림은 4가지 청킹 전략의 특성과 성능을 비교한 것입니다.

Recursive Character Splitting이 비용 대비 성능에서 가장 균형 잡힌 선택입니다. 소규모 기업이라면 여기서 시작하고, 검색 품질 평가 결과에 따라 시맨틱이나 계층적 청킹으로 전환하는 순서가 맞습니다. 각 전략을 구체적으로 살펴보겠습니다.

고정 크기 청킹 - 가장 단순하지만 가장 위험합니다

고정 크기 청킹은 문서를 일정한 문자 수 또는 토큰 수로 기계적으로 분할합니다.

복사
# 고정 크기 청킹 (비권장)
def fixed_size_chunk(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start = end - overlap
    return chunks

이 방식의 문제는 1편에서 충분히 설명했습니다. "3분기 매출은 전년 대비"와 "15% 증가했습니다"로 나뉘면 각각은 의미가 없습니다. CDC 정책 문서 연구에서 고정 크기 청킹의 faithfulness가 0.47~0.51이었다는 사실을 다시 떠올려 보면, 이 방식을 프로덕션에 쓰는 것은 동전 던지기와 비슷합니다.

유일한 장점은 구현이 단순하다는 것입니다. 그러나 이 단순함의 대가가 너무 큽니다.

Recursive Character Splitting - 권장 기본값

LangChain의 RecursiveCharacterTextSplitter가 대표적인 구현체입니다. 이 방식은 문서의 자연스러운 구분자를 순차적으로 시도합니다. 먼저 단락 구분(\n\n)으로 나누고, 그래도 청크가 너무 크면 줄바꿈(\n)으로 나누고, 그래도 크면 문장 구분(. )으로 나누고, 마지막으로 공백( )으로 나눕니다.

복사
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 권장 설정: 400-512 토큰, 50-100 토큰 오버랩
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 약 400-512 토큰에 해당
    chunk_overlap=50,      # 문맥 연결을 위한 오버랩
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len,
)

chunks = splitter.split_text(document_text)

Firecrawl의 2025년 벤치마크에서 이 방식은 85~90%의 recall을 기록했습니다. 고정 크기 대비 눈에 띄게 나아졌으면서도 구현이 간단합니다. 청크 크기를 400~512 토큰으로 설정하는 것이 현재 가장 널리 권장되는 기본값입니다.

청크 크기 선택에 대해 조금 더 구체적으로 이야기하겠습니다. Firecrawl의 벤치마크 가이드에 따르면 쿼리 유형에 따라 최적 크기가 다릅니다. 팩트 기반 질문("Q3 매출이 얼마입니까?")에는 256~512 토큰이 적합하고, 분석 질문("지난 1년간 매출 추세를 분석해 주세요")에는 1024 토큰 이상이 적합합니다. 사내 문서 검색에서는 팩트 질문이 더 빈번하므로 400~512 토큰으로 시작하는 것이 합리적입니다.

오버랩에 대해서는 최근 연구 결과가 엇갈립니다. 일부 연구에서 "오버랩은 측정 가능한 이점 없이 인덱싱 비용만 증가시킨다"는 결과가 나왔습니다. 그러나 문맥이 끊기는 경계 부분에서 오버랩이 도움이 되는 경우도 분명히 있습니다. Firecrawl은 50~100 토큰 오버랩으로 시작한 뒤, 대표 쿼리로 테스트해보고 효과가 없으면 제거하는 접근을 권장합니다.

시맨틱 청킹 - 의미 단위로 자릅니다

시맨틱 청킹은 텍스트의 의미적 일관성을 기준으로 분할합니다. 연속된 문장 간의 임베딩 유사도를 계산하고, 유사도가 급격히 떨어지는 지점을 청크 경계로 삼습니다.

복사
from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.embeddings import OllamaEmbeddings

# Ollama의 로컬 임베딩 모델 사용
embeddings = OllamaEmbeddings(model="nomic-embed-text")

semantic_splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95,  # 유사도 하위 5%를 경계로 사용
)

chunks = semantic_splitter.split_text(document_text)

시맨틱 청킹의 벤치마크 결과는 꽤 좋습니다. Firecrawl에 따르면 91~92%의 recall을 기록했고, Recursive Character Splitting(85~90%) 대비 최대 9% 올라갑니다.

그런데 NAACL 2025 Findings에 게재된 연구가 찬물을 끼얹었습니다. 이 연구에서 고정 200단어 청크가 시맨틱 청킹과 동등하거나 더 나은 성능을 보인 겁니다. 비용은 시맨틱 청킹이 훨씬 높습니다. 모든 문장 쌍에 대해 임베딩을 계산해야 하기 때문입니다.

그렇다면 시맨틱 청킹은 쓸모없는 건가요? 그렇지 않습니다. 같은 NAACL 논문에서 임상 의사결정 지원 도메인을 테스트했을 때, 적응형 청킹(시맨틱 청킹의 한 형태)이 87% 정확도를 달성한 반면 고정 크기 청킹은 13%에 그쳤습니다. 도메인에 따라 차이가 극적으로 달라진다는 뜻입니다.

정리하면 이렇습니다. 일반적인 사내 문서에서는 Recursive Character Splitting이 비용 대비 성능이 충분합니다. 의료, 법률, 금융 같은 전문 도메인에서 정밀도가 생명이라면 시맨틱 청킹에 투자할 가치가 있고요. 소규모 기업이라면 Recursive Character Splitting으로 시작하고, 검색 품질 평가(3편에서 다룰 RAGAS)에서 문제가 보이면 그때 시맨틱 청킹으로 전환하는 순서가 맞습니다.

계층적 청킹 - 부모-자식 구조

Databricks의 기술 블로그에서 소개된 계층적 청킹은 문서를 다단계로 분할합니다. 문서 전체를 대섹션으로 나누고, 각 대섹션을 하위 섹션으로 나누고, 각 하위 섹션을 단락으로 나눕니다. 검색은 가장 작은 단위(단락)로 수행하되, 결과를 반환할 때는 부모 컨텍스트를 함께 제공합니다.

복사
from dataclasses import dataclass

@dataclass
class HierarchicalChunk:
    """부모-자식 관계를 가진 청크."""
    content: str
    level: int            # 0: 문서, 1: 섹션, 2: 하위섹션, 3: 단락
    parent_id: str = ""
    chunk_id: str = ""
    metadata: dict = None

    def __post_init__(self):
        if self.metadata is None:
            self.metadata = {}

class HierarchicalChunker:
    """마크다운 문서를 계층적으로 청킹합니다."""

    def chunk(self, text: str, doc_id: str = "doc") -> list[HierarchicalChunk]:
        chunks = []
        # 최상위: H1/H2 기준으로 섹션 분할
        sections = self._split_by_headers(text, level=1)

        for sec_idx, section in enumerate(sections):
            sec_id = f"{doc_id}_s{sec_idx}"
            sec_chunk = HierarchicalChunk(
                content=section["content"],
                level=1,
                parent_id=doc_id,
                chunk_id=sec_id,
                metadata={"heading": section.get("heading", "")},
            )
            chunks.append(sec_chunk)

            # 하위 섹션: H3 기준으로 분할
            subsections = self._split_by_headers(section["content"], level=2)
            for sub_idx, subsection in enumerate(subsections):
                sub_id = f"{sec_id}_ss{sub_idx}"
                sub_chunk = HierarchicalChunk(
                    content=subsection["content"],
                    level=2,
                    parent_id=sec_id,
                    chunk_id=sub_id,
                    metadata={"heading": subsection.get("heading", "")},
                )
                chunks.append(sub_chunk)

                # 단락 수준 분할
                paragraphs = self._split_paragraphs(subsection["content"])
                for para_idx, para in enumerate(paragraphs):
                    para_id = f"{sub_id}_p{para_idx}"
                    para_chunk = HierarchicalChunk(
                        content=para,
                        level=3,
                        parent_id=sub_id,
                        chunk_id=para_id,
                    )
                    chunks.append(para_chunk)

        return chunks

    def _split_by_headers(self, text: str, level: int) -> list[dict]:
        """마크다운 헤더 기준으로 분할합니다."""
        import re
        if level == 1:
            pattern = r'^#{1,2}\s+(.+)$'
        else:
            pattern = r'^#{3}\s+(.+)$'

        sections = []
        current_heading = ""
        current_lines = []

        for line in text.split("\n"):
            if re.match(pattern, line, re.MULTILINE):
                if current_lines:
                    sections.append({
                        "heading": current_heading,
                        "content": "\n".join(current_lines),
                    })
                current_heading = line.strip("# ").strip()
                current_lines = [line]
            else:
                current_lines.append(line)

        if current_lines:
            sections.append({
                "heading": current_heading,
                "content": "\n".join(current_lines),
            })

        return sections if sections else [{"heading": "", "content": text}]

    def _split_paragraphs(self, text: str) -> list[str]:
        """빈 줄 기준으로 단락을 분할합니다."""
        paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
        # 너무 짧은 단락은 이전 단락에 병합
        merged = []
        for para in paragraphs:
            if merged and len(para) < 100:
                merged[-1] += "\n\n" + para
            else:
                merged.append(para)
        return merged

이 방식은 검색 정밀도와 컨텍스트를 동시에 챙길 수 있습니다. "프로젝트 알파의 3분기 매출"을 검색하면 해당 단락(level 3)이 히트되고, LLM에는 그 단락이 속한 섹션(level 1)의 컨텍스트까지 함께 전달할 수 있습니다. LlamaIndex에서는 이를 "Parent Document Retriever" 패턴으로 구현하고 있습니다.

구조화된 문서(기술 문서, 정책 문서, 매뉴얼)에 적합합니다. 비구조화된 자유형 텍스트에는 헤더 기반 분할이 어려우므로 Recursive Character Splitting이 더 나은 선택입니다.

청킹 전략 선택 가이드

어떤 전략을 써야 할지 결정하는 프레임워크를 정리합니다.

Firecrawl의 벤치마크 가이드가 제안하는 접근법이 실용적입니다. RecursiveCharacterTextSplitter 400~512 토큰으로 시작하고, 대표 문서 50~100개와 실제 쿼리 20~30개로 2~3가지 전략을 비교 테스트합니다. 평가 지표로는 3편에서 다룰 RAGAS의 Context Precision과 Context Recall을 사용합니다.

문서 유형별로 정리하면 이렇습니다. 사내 규정, 정책 문서, 기술 매뉴얼처럼 구조가 명확한 문서에는 계층적 청킹이 잘 맞습니다. 이메일, 회의록, 자유형 보고서에는 Recursive Character Splitting이 낫고요. 의료나 법률, 금융처럼 전문 용어가 밀집된 도메인에서는 시맨틱 청킹을 고려할 만합니다. 소량이지만 틀리면 안 되는 문서(계약서, 인증 문서 등)에는 LLM 기반 청킹까지 투자할 가치가 있습니다.

Contextual Retrieval - 문맥을 살리는 기법

어떤 청킹 전략을 쓰든 피할 수 없는 한계가 있습니다. 청크는 전체 문서에서 잘려 나온 조각이므로, 원래 문맥의 일부가 사라집니다. "이 회사의 3분기 매출이 증가한 것은 신제품 출시 효과 때문입니다"라는 문장이 있을 때, 이 문장만 따로 떼어놓으면 "이 회사"가 어디인지, "신제품"이 무엇인지 알 수 없습니다.

Anthropic이 2024년에 발표한 Contextual Retrieval이 이 문제를 직접적으로 해결합니다. 아이디어는 단순합니다. 각 청크를 임베딩하기 전에, LLM에게 "이 청크가 전체 문서에서 어떤 맥락에 있는지" 짧게 설명해달라고 요청하고, 그 설명을 청크 앞에 붙이는 겁니다.

다음 그림은 Contextual Retrieval의 작동 원리와 검색 실패율 감소 효과를 보여줍니다.

전체 문서와 개별 청크를 LLM에 함께 보내면 짧은 문맥 설명이 생성되고, 이것을 청크 앞에 붙여 풍부해진 청크를 만듭니다. BM25와 리랭킹까지 결합하면 검색 실패율이 5.7%에서 1.9%로 67% 감소합니다.

Anthropic 공식 문서에서 제시한 프롬프트 템플릿은 다음과 같습니다.

복사
<document>
{{WHOLE_DOCUMENT}}
</document>

Here is the chunk we want to situate within the whole document
<chunk>
{{CHUNK_CONTENT}}
</chunk>

Please give a short succinct context to situate this chunk within the
overall document for the purposes of improving search retrieval of
the chunk. Answer only with the succinct context and nothing else.

전체 문서와 개별 청크를 함께 보내면, LLM이 50~100 토큰 정도의 컨텍스트를 생성합니다. 예를 들어 "이 청크는 삼성전자의 2025년 3분기 실적 보고서에서 반도체 사업부의 매출 성장에 대해 설명하는 부분입니다"라는 문맥이 청크 앞에 추가됩니다.

Anthropic의 벤치마크 결과를 보면 수치가 확실합니다. Contextual Embeddings만 적용했을 때 검색 실패율이 35% 감소(5.7%에서 3.7%로)했습니다. BM25 키워드 검색을 더하면 49% 감소(5.7%에서 2.9%로)했습니다. 리랭킹까지 추가하면 67% 감소(5.7%에서 1.9%로)했습니다.

이 수치가 뜻하는 바를 풀어보겠습니다. 기존에 100번 검색하면 5.7번은 관련 문서를 못 찾았는데, Contextual Retrieval + BM25 + 리랭킹을 적용하면 1.9번으로 줄어든다는 겁니다. 검색이 정확해지면 LLM이 올바른 컨텍스트를 받으니, 최종 답변 품질도 자연스럽게 올라갑니다.

비용은 어떨까요? Anthropic 문서에 따르면 프롬프트 캐싱을 활용할 경우 100만 문서 토큰당 약 $1.02의 일회성 비용이 발생합니다. 문서가 변경되지 않으면 다시 처리할 필요가 없으니, 초기 인덱싱 비용으로 생각하면 됩니다. Anthropic은 20만 토큰 이상의 지식 베이스에서 경제적으로 가치 있다고 명시하고 있습니다. A4 150~200페이지 이상이면 투자 대비 효과가 있다는 뜻입니다.

구현 코드를 보겠습니다.

복사
# contextual_chunker.py
import json
from dataclasses import dataclass

@dataclass
class ContextualChunk:
    """문맥 정보가 추가된 청크."""
    original_content: str
    context: str
    contextualized_content: str  # context + original_content
    chunk_index: int
    metadata: dict = None

    def __post_init__(self):
        if self.metadata is None:
            self.metadata = {}

class ContextualChunker:
    """Anthropic Contextual Retrieval 방식으로 청크에 문맥을 추가합니다."""

    def __init__(self, llm_client, base_chunker):
        """
        llm_client: LLM API 호출 객체 (generate 메서드 필요)
        base_chunker: 기본 청킹을 수행할 객체 (split_text 메서드 필요)
        """
        self.llm = llm_client
        self.base_chunker = base_chunker

    def chunk_with_context(self, document: str, metadata: dict = None) -> list[ContextualChunk]:
        """문서를 청킹하고 각 청크에 문맥 정보를 추가합니다."""
        # 1단계: 기본 청킹
        raw_chunks = self.base_chunker.split_text(document)
        print(f"기본 청킹 완료: {len(raw_chunks)}개 청크")

        # 2단계: 각 청크에 문맥 추가
        contextual_chunks = []
        for idx, chunk in enumerate(raw_chunks):
            context = self._generate_context(document, chunk)
            contextualized = f"{context}\n\n{chunk}"

            contextual_chunks.append(ContextualChunk(
                original_content=chunk,
                context=context,
                contextualized_content=contextualized,
                chunk_index=idx,
                metadata=metadata or {},
            ))

            if (idx + 1) % 10 == 0:
                print(f"문맥 추가 진행: {idx + 1}/{len(raw_chunks)}")

        print(f"문맥 추가 완료: {len(contextual_chunks)}개")
        return contextual_chunks

    def _generate_context(self, document: str, chunk: str) -> str:
        """LLM을 사용하여 청크의 문맥 정보를 생성합니다."""
        prompt = f"""<document>
{document}
</document>

Here is the chunk we want to situate within the whole document
<chunk>
{chunk}
</chunk>

Please give a short succinct context to situate this chunk within the
overall document for the purposes of improving search retrieval of
the chunk. Answer only with the succinct context and nothing else."""

        response = self.llm.generate(prompt, max_tokens=150)
        return response.strip()

이 코드에서 llm_client는 Ollama 로컬 모델이든 Claude API든 상관없습니다. 문맥 생성은 고도의 추론이 필요하지 않으므로 비용이 낮은 모델을 쓰는 것이 합리적입니다. Ollama에서 llama3.2 같은 경량 모델을 쓰면 비용이 사실상 무료입니다.

한 가지 주의할 점이 있습니다. 전체 문서를 매 청크마다 프롬프트에 포함해야 하므로, 문서가 길면 LLM 호출 비용이 높아집니다. Anthropic의 프롬프트 캐싱을 쓸 수 있다면 같은 문서에 대한 반복 호출 비용이 크게 줄어듭니다. 로컬 모델을 쓸 경우 비용 문제는 없지만, 문서가 모델의 컨텍스트 윈도우를 초과하면 문서를 요약하거나 분할해서 보내야 합니다.

Ollama를 활용한 구현체는 다음과 같습니다.

복사
# ollama_llm_client.py
import requests

class OllamaLLMClient:
    """Ollama 로컬 LLM 클라이언트."""

    def __init__(self, model: str = "llama3.2", base_url: str = "http://localhost:11434"):
        self.model = model
        self.base_url = base_url

    def generate(self, prompt: str, max_tokens: int = 150) -> str:
        response = requests.post(
            f"{self.base_url}/api/generate",
            json={
                "model": self.model,
                "prompt": prompt,
                "stream": False,
                "options": {"num_predict": max_tokens},
            },
        )
        response.raise_for_status()
        return response.json()["response"]

Contextual Retrieval을 적용할지는 문서 규모와 예산으로 결정합니다. 문서가 수백 개 미만이고 검색 품질이 이미 괜찮다면 굳이 적용할 필요가 없습니다. 문서가 수천 개 이상이고 검색 정확도가 중요하다면 적용할 가치가 있고요. 로컬 LLM을 쓸 수 있다면 비용 부담은 거의 없습니다.

임베딩 모델 선택 - 숨겨진 비용 구조

청킹된 텍스트를 벡터 검색이 가능하도록 임베딩 벡터로 변환해야 합니다. 어떤 임베딩 모델을 쓰느냐에 따라 검색 품질과 운영 비용이 달라집니다.

2026년 기준으로 세 가지 선택지가 있습니다. OpenAI/Voyage AI 같은 상용 API, Ollama 기반 로컬 모델, 그리고 직접 파인튜닝한 모델입니다. 소규모 기업에 세 번째 선택은 현실적이지 않으므로 앞의 두 가지를 비교하겠습니다.

OpenAI의 text-embedding-3-large는 100만 토큰당 $0.130입니다. 매일 1,000건의 문서를 처리하고 각 문서가 평균 2,000 토큰이라면, 하루 200만 토큰을 소비하고 일일 비용은 $0.26입니다. 월 기준 $7.80, 쿼리 임베딩 비용까지 합치면 월 $10~15 정도입니다. 절대 금액으로는 부담이 크지 않습니다.

그런데 숨겨진 비용이 있습니다. 문서가 업데이트되면 임베딩을 다시 생성해야 합니다. Contextual Retrieval을 적용하면 문맥 생성 비용도 추가됩니다. 인덱스를 재구축해야 하는 상황(임베딩 모델 변경, 청킹 전략 변경)이 오면 전체 문서를 다시 처리해야 합니다. 이런 상황이 반복되면 비용이 예상 외로 높아집니다.

Ollama 기반 로컬 임베딩은 이 문제를 아예 없앱니다. Ollama 공식 문서에 따르면 nomic-embed-text는 OpenAI의 text-embedding-ada-002와 text-embedding-3-small보다 성능이 우수하면서, 한 번 설치하면 추가 비용이 없습니다. mxbai-embed-large는 text-embedding-3-large보다 성능이 좋으면서 크기가 작습니다.

복사
# embedding_service.py
import requests
import numpy as np
from abc import ABC, abstractmethod

class EmbeddingService(ABC):
    """임베딩 서비스 추상 인터페이스."""

    @abstractmethod
    def embed(self, text: str) -> list[float]:
        """단일 텍스트를 임베딩합니다."""
        pass

    @abstractmethod
    def embed_batch(self, texts: list[str]) -> list[list[float]]:
        """여러 텍스트를 배치로 임베딩합니다."""
        pass

class OllamaEmbedding(EmbeddingService):
    """Ollama 로컬 임베딩 서비스."""

    def __init__(self, model: str = "nomic-embed-text", base_url: str = "http://localhost:11434"):
        self.model = model
        self.base_url = base_url

    def embed(self, text: str) -> list[float]:
        response = requests.post(
            f"{self.base_url}/api/embed",
            json={"model": self.model, "input": text},
        )
        response.raise_for_status()
        return response.json()["embeddings"][0]

    def embed_batch(self, texts: list[str]) -> list[list[float]]:
        response = requests.post(
            f"{self.base_url}/api/embed",
            json={"model": self.model, "input": texts},
        )
        response.raise_for_status()
        return response.json()["embeddings"]

class OpenAIEmbedding(EmbeddingService):
    """OpenAI 임베딩 서비스. API 비용이 발생합니다."""

    def __init__(self, api_key: str, model: str = "text-embedding-3-small"):
        from openai import OpenAI
        self.client = OpenAI(api_key=api_key)
        self.model = model

    def embed(self, text: str) -> list[float]:
        response = self.client.embeddings.create(
            model=self.model,
            input=text,
        )
        return response.data[0].embedding

    def embed_batch(self, texts: list[str]) -> list[list[float]]:
        response = self.client.embeddings.create(
            model=self.model,
            input=texts,
        )
        return [item.embedding for item in response.data]

추상 인터페이스를 쓰는 이유가 있습니다. RAG 시스템을 운영하다 보면 임베딩 모델을 교체해야 하는 시점이 옵니다. 더 좋은 모델이 나왔거나, 비용 구조를 바꿔야 하거나, 다국어 지원이 필요해지거나. 이때 인터페이스가 통일되어 있으면 구현체만 교체하면 되거든요. 2025년 RAG 생태계에서 팩토리 기반이나 어댑터 기반 추상화가 표준이 된 것도 이런 이유입니다.

모델 선택을 정리하면 이렇습니다. 비용이 최우선이라면 Ollama nomic-embed-text로 시작합니다. 검색 품질이 아쉬우면 mxbai-embed-large로 올리고요. 다국어 문서가 많거나 특정 도메인에서 정밀도가 중요하면 OpenAI text-embedding-3-large나 Voyage AI를 검토합니다.

한 가지 주의할 점이 있습니다. 위에서 언급한 벤치마크 수치는 영어 기준입니다. 한국어 사내 문서를 처리하는 경우, nomic-embed-text는 다국어를 지원하지만 영어에 최적화된 모델이므로 한국어 검색 품질이 기대에 미치지 못할 수 있습니다. 한국어 문서 비중이 높다면 multilingual-e5-large, bge-m3, 또는 Ollama에서 ollama pull snowflake-arctic-embed2로 설치할 수 있는 다국어 특화 임베딩 모델을 먼저 검토하는 것이 좋습니다. 대표 쿼리 20~30개를 준비해서 영어 모델과 다국어 모델의 검색 결과를 비교한 뒤 결정하면 가장 확실합니다.

비용 테이블을 만들어보면 이렇습니다.

Ollama nomic-embed-text는 초기 GPU 구매나 기존 장비를 활용하면 쿼리당 추가 비용이 없습니다. 768차원 벡터를 생성합니다. Ollama mxbai-embed-large는 같은 조건에서 1024차원 벡터를 생성하고, 정밀도가 조금 더 높습니다. OpenAI text-embedding-3-small은 100만 토큰당 $0.020이고 1536차원입니다. OpenAI text-embedding-3-large는 100만 토큰당 $0.130이고 3072차원입니다.

소규모 기업의 현실적 권장 구성은 Ollama nomic-embed-text입니다. GPU가 없는 일반 노트북이나 서버에서도 CPU 모드로 실행할 수 있습니다. 속도는 GPU 대비 느리지만, 인덱싱은 오프라인 배치로 처리하면 되므로 실시간 속도가 필수는 아닙니다. 쿼리 임베딩은 단일 텍스트이므로 CPU로도 100ms 이내에 처리 가능합니다.

벡터 저장소 - 어디에 담을 것인가

임베딩이 생성되면 이를 저장하고 검색할 벡터 저장소가 필요합니다. 2026년 기준으로 소규모 기업이 현실적으로 고를 수 있는 선택지를 보겠습니다.

ChromaDB는 가장 간단합니다. pip install chromadb 한 줄이면 설치가 끝나고, 파일 기반으로 작동하며, 별도 서버 없이 애플리케이션에 내장할 수 있습니다. 문서 10만 건 이하 규모에 적합하고, 오픈소스라 비용도 없습니다.

pgvector는 PostgreSQL을 이미 쓰고 있는 기업에 맞습니다. 확장만 추가하면 벡터 검색이 되거든요. 기존 인프라를 그대로 활용하니 추가 비용이 거의 없고, SQL과 벡터 검색을 하나의 쿼리로 결합할 수 있습니다. "인사팀 문서 중에서 퇴직 관련 내용을 검색"처럼 메타데이터 필터와 벡터 검색을 동시에 걸 수 있습니다.

Milvus는 수십억 벡터를 처리할 수 있는 고성능 벡터 DB인데, 소규모 기업에는 과합니다. 문서가 수십만 건 이상으로 성장할 계획이 확실한 경우에만 고려하면 됩니다.

ChromaDB를 사용한 벡터 저장 및 검색 코드입니다.

복사
# vector_store.py
import chromadb
from chromadb.config import Settings

class VectorStore:
    """ChromaDB 기반 벡터 저장소."""

    def __init__(self, persist_dir: str = "./chroma_db", collection_name: str = "documents"):
        self.client = chromadb.PersistentClient(
            path=persist_dir,
            settings=Settings(anonymized_telemetry=False),
        )
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"},  # 코사인 유사도 사용
        )

    def add_chunks(
        self,
        chunks: list[ContextualChunk],
        embeddings: list[list[float]],
        doc_metadata: dict = None,
    ):
        """청크와 임베딩을 저장합니다."""
        ids = []
        documents = []
        metadatas = []
        embedding_list = []

        for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
            chunk_id = f"chunk_{chunk.chunk_index}_{hash(chunk.original_content) % 100000}"
            ids.append(chunk_id)
            documents.append(chunk.contextualized_content)
            embedding_list.append(embedding)

            meta = {**chunk.metadata}
            if doc_metadata:
                meta.update(doc_metadata)
            meta["chunk_index"] = chunk.chunk_index
            meta["has_context"] = bool(chunk.context)
            metadatas.append(meta)

        # ChromaDB는 배치 크기 제한이 있으므로 분할 처리
        batch_size = 500
        for start in range(0, len(ids), batch_size):
            end = start + batch_size
            self.collection.add(
                ids=ids[start:end],
                documents=documents[start:end],
                embeddings=embedding_list[start:end],
                metadatas=metadatas[start:end],
            )

        print(f"{len(ids)}개 청크 저장 완료")

    def search(
        self,
        query_embedding: list[float],
        n_results: int = 10,
        where: dict = None,
    ) -> list[dict]:
        """벡터 유사도 검색을 수행합니다."""
        kwargs = {
            "query_embeddings": [query_embedding],
            "n_results": n_results,
        }
        if where:
            kwargs["where"] = where

        results = self.collection.query(**kwargs)

        search_results = []
        for i in range(len(results["ids"][0])):
            search_results.append({
                "id": results["ids"][0][i],
                "content": results["documents"][0][i],
                "metadata": results["metadatas"][0][i],
                "distance": results["distances"][0][i],
            })

        return search_results

    def get_stats(self) -> dict:
        """저장소 통계를 반환합니다."""
        return {
            "total_chunks": self.collection.count(),
            "collection_name": self.collection.name,
        }

pgvector를 사용하는 경우의 코드도 제공합니다. 이미 PostgreSQL을 운영 중인 기업이라면 이쪽이 더 자연스럽습니다.

복사
# pgvector_store.py
import psycopg2
from pgvector.psycopg2 import register_vector

class PgVectorStore:
    """pgvector 기반 벡터 저장소."""

    def __init__(self, connection_string: str, table_name: str = "document_chunks"):
        self.conn = psycopg2.connect(connection_string)
        self.table_name = table_name
        register_vector(self.conn)
        self._init_table()

    def _init_table(self):
        """테이블이 없으면 생성합니다."""
        with self.conn.cursor() as cur:
            cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
            cur.execute(f"""
                CREATE TABLE IF NOT EXISTS {self.table_name} (
                    id SERIAL PRIMARY KEY,
                    content TEXT NOT NULL,
                    context TEXT,
                    embedding vector(768),
                    metadata JSONB DEFAULT '{{}}'::jsonb,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)
            # HNSW 인덱스 생성 (코사인 유사도)
            cur.execute(f"""
                CREATE INDEX IF NOT EXISTS idx_{self.table_name}_embedding
                ON {self.table_name}
                USING hnsw (embedding vector_cosine_ops)
                WITH (m = 16, ef_construction = 64)
            """)
            self.conn.commit()

    def add_chunk(self, content: str, context: str, embedding: list[float], metadata: dict = None):
        """단일 청크를 저장합니다."""
        import json
        with self.conn.cursor() as cur:
            cur.execute(
                f"""INSERT INTO {self.table_name} (content, context, embedding, metadata)
                    VALUES (%s, %s, %s::vector, %s::jsonb)""",
                (content, context, embedding, json.dumps(metadata or {})),
            )
            self.conn.commit()

    def search(self, query_embedding: list[float], n_results: int = 10, department: str = None) -> list[dict]:
        """벡터 유사도 검색. 메타데이터 필터링 지원."""
        with self.conn.cursor() as cur:
            # pgvector에서 <=>는 코사인 거리
            if department:
                cur.execute(
                    f"""SELECT id, content, context, metadata,
                               1 - (embedding <=> %s::vector) AS similarity
                        FROM {self.table_name}
                        WHERE metadata->>'department' = %s
                        ORDER BY embedding <=> %s::vector
                        LIMIT %s""",
                    [query_embedding, department, query_embedding, n_results],
                )
            else:
                cur.execute(
                    f"""SELECT id, content, context, metadata,
                               1 - (embedding <=> %s::vector) AS similarity
                        FROM {self.table_name}
                        ORDER BY embedding <=> %s::vector
                        LIMIT %s""",
                    [query_embedding, query_embedding, n_results],
                )

            results = []
            for row in cur.fetchall():
                results.append({
                    "id": row[0],
                    "content": row[1],
                    "context": row[2],
                    "metadata": row[3],
                    "similarity": float(row[4]),
                })

            return results

pgvector의 장점은 메타데이터 필터링입니다. WHERE metadata->>'department' = '인사팀' 같은 SQL 조건을 벡터 검색과 함께 걸 수 있습니다. ChromaDB도 where 파라미터로 필터링을 지원하지만, 복잡한 조건에서는 PostgreSQL의 JSONB 인덱싱이 더 유연합니다.

지식 그래프 구축 - 선택적이지만 강력합니다

1편에서 온톨로지와 지식 그래프를 설명했습니다. "프로젝트 알파 담당자가 누구입니까?" 같은 관계 기반 질문에 벡터 검색만으로는 한계가 있다는 것을 봤습니다. 지식 그래프는 이런 관계형 질문의 정확도를 20~35% 향상시킵니다.

그렇다고 모든 RAG 시스템에 지식 그래프가 필요하지는 않습니다. 단순 팩트 검색이 대부분이라면 벡터 검색으로 충분하거든요. 지식 그래프가 빛을 발하는 상황은 구체적입니다. "이 프로젝트에 관련된 모든 사람을 알려주세요", "A 정책과 B 정책이 충돌하는 부분이 있습니까?", "이 기술 스택을 사용하는 다른 프로젝트는 무엇입니까?" 같은, 여러 엔티티 간의 관계를 탐색해야 하는 질문입니다.

소규모 기업이 지식 그래프를 구축한다면, Microsoft GraphRAG보다 LightRAG가 현실적인 선택입니다. LightRAG는 홍콩대학교에서 개발한 경량 그래프 기반 RAG로, GitHub 스타 14.6K를 기록하고 있습니다. GraphRAG 대비 설정이 단순하고, 리소스 요구사항이 낮습니다. 관계형 QA에서 GraphRAG보다 최대 10% 낮은 정확도를 보이지만, 비용 민감 환경에서는 충분한 성능입니다.

LLM을 사용해 문서에서 엔티티와 관계를 추출하는 기본 구현을 보겠습니다.

복사
# knowledge_graph.py
import json
from dataclasses import dataclass, field

@dataclass
class Entity:
    """지식 그래프의 노드."""
    name: str
    entity_type: str  # person, project, technology, policy 등
    properties: dict = field(default_factory=dict)

@dataclass
class Relationship:
    """지식 그래프의 엣지."""
    source: str
    target: str
    relation_type: str  # works_on, uses, depends_on, authored_by 등
    properties: dict = field(default_factory=dict)

class SimpleKnowledgeGraph:
    """파일 기반 지식 그래프. Neo4j 없이 동작합니다."""

    def __init__(self):
        self.entities: dict[str, Entity] = {}
        self.relationships: list[Relationship] = []

    def add_entity(self, entity: Entity):
        """엔티티를 추가하거나 기존 엔티티의 속성을 병합합니다."""
        key = f"{entity.entity_type}:{entity.name}".lower()
        if key in self.entities:
            self.entities[key].properties.update(entity.properties)
        else:
            self.entities[key] = entity

    def add_relationship(self, rel: Relationship):
        """관계를 추가합니다."""
        self.relationships.append(rel)

    def find_related(self, entity_name: str, max_depth: int = 2) -> list[dict]:
        """주어진 엔티티와 관련된 모든 엔티티를 찾습니다."""
        entity_name_lower = entity_name.lower()
        visited = set()
        results = []

        def traverse(name: str, depth: int):
            if depth > max_depth or name in visited:
                return
            visited.add(name)

            for rel in self.relationships:
                if rel.source.lower() == name:
                    results.append({
                        "from": rel.source,
                        "relation": rel.relation_type,
                        "to": rel.target,
                        "depth": depth,
                    })
                    traverse(rel.target.lower(), depth + 1)
                elif rel.target.lower() == name:
                    results.append({
                        "from": rel.target,
                        "relation": f"reverse_{rel.relation_type}",
                        "to": rel.source,
                        "depth": depth,
                    })
                    traverse(rel.source.lower(), depth + 1)

        traverse(entity_name_lower, 0)
        return results

    def save(self, file_path: str):
        """그래프를 JSON 파일로 저장합니다."""
        data = {
            "entities": {
                k: {"name": v.name, "type": v.entity_type, "properties": v.properties}
                for k, v in self.entities.items()
            },
            "relationships": [
                {
                    "source": r.source,
                    "target": r.target,
                    "type": r.relation_type,
                    "properties": r.properties,
                }
                for r in self.relationships
            ],
        }
        with open(file_path, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    def load(self, file_path: str):
        """JSON 파일에서 그래프를 로드합니다."""
        with open(file_path, "r", encoding="utf-8") as f:
            data = json.load(f)

        for key, val in data["entities"].items():
            self.entities[key] = Entity(
                name=val["name"],
                entity_type=val["type"],
                properties=val.get("properties", {}),
            )

        for rel in data["relationships"]:
            self.relationships.append(Relationship(
                source=rel["source"],
                target=rel["target"],
                relation_type=rel["type"],
                properties=rel.get("properties", {}),
            ))

class EntityExtractor:
    """LLM을 사용해 텍스트에서 엔티티와 관계를 추출합니다."""

    def __init__(self, llm_client):
        self.llm = llm_client

    def extract(self, text: str) -> tuple[list[Entity], list[Relationship]]:
        """텍스트에서 엔티티와 관계를 추출합니다."""
        prompt = f"""다음 텍스트에서 엔티티(사람, 프로젝트, 기술, 정책, 부서, 회사)와
그들 사이의 관계를 추출해주세요.

텍스트:
{text}

다음 JSON 형식으로 응답해주세요:
{{
  "entities": [
    {{"name": "엔티티명", "type": "person|project|technology|policy|department|company", "properties": {{}}}}
  ],
  "relationships": [
    {{"source": "엔티티명1", "target": "엔티티명2", "type": "works_on|uses|depends_on|authored_by|belongs_to|manages"}}
  ]
}}

JSON만 응답하고 다른 텍스트는 포함하지 마세요."""

        response = self.llm.generate(prompt, max_tokens=1000)

        try:
            # JSON 파싱 시도
            data = json.loads(response)
            entities = [
                Entity(name=e["name"], entity_type=e["type"], properties=e.get("properties", {}))
                for e in data.get("entities", [])
            ]
            relationships = [
                Relationship(source=r["source"], target=r["target"], relation_type=r["type"])
                for r in data.get("relationships", [])
            ]
            return entities, relationships
        except (json.JSONDecodeError, KeyError) as e:
            print(f"[WARN] 엔티티 추출 실패: {e}")
            return [], []

이 코드는 Neo4j 없이 순수 Python과 JSON 파일로 지식 그래프를 돌립니다. 문서 수천 건 규모에서는 이 방식으로 충분하고요. 문서가 수만 건 이상으로 불어나거나, 실시간 그래프 쿼리가 필요해지면 그때 Neo4j를 도입하면 됩니다.

엔티티 해상도에 대해 한 가지 주의할 점이 있습니다. 1편에서 언급했듯이 엔티티 해상도(같은 개념이 다른 이름으로 언급된 것을 통합하는 작업)의 정확도가 85% 이하이면 전체 시스템이 불안정해집니다. "김철수", "철수", "Kim CS", "K.C. Kim"이 모두 같은 사람을 가리키는 것을 시스템이 인식해야 합니다. 이를 위해 추출된 엔티티를 수동으로 한 번 검토하고, 동의어 사전을 관리하는 것이 현실적인 접근법입니다.

복사
# entity_resolver.py

class EntityResolver:
    """엔티티 이름 통합을 관리합니다."""

    def __init__(self, alias_file: str = "entity_aliases.json"):
        self.alias_file = alias_file
        self.aliases: dict[str, str] = {}  # 별칭 -> 표준 이름
        self._load_aliases()

    def _load_aliases(self):
        try:
            with open(self.alias_file, "r", encoding="utf-8") as f:
                self.aliases = json.load(f)
        except FileNotFoundError:
            self.aliases = {}

    def resolve(self, name: str) -> str:
        """별칭을 표준 이름으로 변환합니다."""
        return self.aliases.get(name.lower(), name)

    def add_alias(self, alias: str, canonical_name: str):
        """별칭을 등록합니다."""
        self.aliases[alias.lower()] = canonical_name
        self._save_aliases()

    def _save_aliases(self):
        with open(self.alias_file, "w", encoding="utf-8") as f:
            json.dump(self.aliases, f, ensure_ascii=False, indent=2)

지식 그래프 구축은 선택 사항입니다. 먼저 벡터 검색과 하이브리드 검색으로 시스템을 돌려보고, 관계형 질문에서 답변 품질이 아쉬우면 그때 지식 그래프를 얹는 순서가 소규모 기업에 맞습니다.

전체 파이프라인 조립

지금까지 만든 구성 요소를 하나의 파이프라인으로 조립합니다. 문서 로딩부터 벡터 저장까지, 하나의 스크립트로 돌릴 수 있게 만듭니다.

복사
# pipeline.py
"""
RAG 데이터 수집 파이프라인.

사용법:
    python pipeline.py --input-dir ./documents --persist-dir ./chroma_db
"""
import argparse
import time
from pathlib import Path

# 위에서 정의한 모듈들을 임포트
from document_loader import DocumentLoader, Document
from metadata_enricher import MetadataEnricher
from contextual_chunker import ContextualChunker, ContextualChunk
from embedding_service import OllamaEmbedding
from vector_store import VectorStore
from knowledge_graph import SimpleKnowledgeGraph, EntityExtractor
from ollama_llm_client import OllamaLLMClient

from langchain.text_splitter import RecursiveCharacterTextSplitter

class RAGPipeline:
    """문서 수집부터 인덱싱까지의 전체 파이프라인."""

    def __init__(
        self,
        persist_dir: str = "./chroma_db",
        use_contextual: bool = True,
        build_knowledge_graph: bool = False,
        embedding_model: str = "nomic-embed-text",
        llm_model: str = "llama3.2",
    ):
        # 구성 요소 초기화
        self.loader = DocumentLoader()
        self.enricher = MetadataEnricher(department_mapping={
            "/hr/": "인사팀",
            "/dev/": "개발팀",
            "/sales/": "영업팀",
            "/finance/": "재무팀",
        })
        self.base_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            separators=["\n\n", "\n", ". ", " ", ""],
        )
        self.embedding = OllamaEmbedding(model=embedding_model)
        self.vector_store = VectorStore(persist_dir=persist_dir)

        self.use_contextual = use_contextual
        self.build_kg = build_knowledge_graph

        if use_contextual or build_knowledge_graph:
            self.llm = OllamaLLMClient(model=llm_model)

        if use_contextual:
            self.contextual_chunker = ContextualChunker(
                llm_client=self.llm,
                base_chunker=self.base_splitter,
            )

        if build_knowledge_graph:
            self.kg = SimpleKnowledgeGraph()
            self.entity_extractor = EntityExtractor(self.llm)

    def process_directory(self, input_dir: str):
        """디렉토리 내 모든 문서를 처리합니다."""
        start_time = time.time()

        # 1. 문서 로딩
        print("=" * 60)
        print("1단계: 문서 로딩")
        print("=" * 60)
        documents = self.loader.load_directory(input_dir)
        if not documents:
            print("처리할 문서가 없습니다.")
            return

        # 2. 메타데이터 추가
        print("\n2단계: 메타데이터 추가")
        for doc in documents:
            self.enricher.enrich(doc)

        # 3. 문서별 처리
        total_chunks = 0
        for doc_idx, doc in enumerate(documents):
            print(f"\n{'=' * 60}")
            print(f"문서 처리 중: [{doc_idx + 1}/{len(documents)}] {doc.metadata.get('file_name', 'unknown')}")
            print(f"{'=' * 60}")

            chunks = self._process_single_document(doc)
            total_chunks += len(chunks)

        # 4. 지식 그래프 저장
        if self.build_kg:
            kg_path = str(Path(input_dir).parent / "knowledge_graph.json")
            self.kg.save(kg_path)
            print(f"\n지식 그래프 저장: {kg_path}")
            print(f"  엔티티: {len(self.kg.entities)}개")
            print(f"  관계: {len(self.kg.relationships)}개")

        elapsed = time.time() - start_time
        print(f"\n{'=' * 60}")
        print(f"파이프라인 완료")
        print(f"  처리 문서: {len(documents)}개")
        print(f"  생성 청크: {total_chunks}개")
        print(f"  소요 시간: {elapsed:.1f}초")
        print(f"{'=' * 60}")

    def _process_single_document(self, doc: Document) -> list:
        """단일 문서를 청킹, 임베딩, 저장합니다."""

        # 청킹
        if self.use_contextual:
            chunks = self.contextual_chunker.chunk_with_context(
                doc.content, metadata=doc.metadata
            )
            texts_to_embed = [c.contextualized_content for c in chunks]
        else:
            raw_chunks = self.base_splitter.split_text(doc.content)
            chunks = [
                ContextualChunk(
                    original_content=chunk,
                    context="",
                    contextualized_content=chunk,
                    chunk_index=i,
                    metadata=doc.metadata,
                )
                for i, chunk in enumerate(raw_chunks)
            ]
            texts_to_embed = [c.original_content for c in chunks]

        print(f"  청크 수: {len(chunks)}")

        # 임베딩 생성 (배치 처리)
        batch_size = 32
        all_embeddings = []
        for i in range(0, len(texts_to_embed), batch_size):
            batch = texts_to_embed[i:i + batch_size]
            embeddings = self.embedding.embed_batch(batch)
            all_embeddings.extend(embeddings)

        print(f"  임베딩 생성 완료: {len(all_embeddings)}개")

        # 벡터 저장소에 저장
        self.vector_store.add_chunks(
            chunks=chunks,
            embeddings=all_embeddings,
            doc_metadata=doc.metadata,
        )

        # 지식 그래프 구축 (선택적)
        if self.build_kg:
            entities, relationships = self.entity_extractor.extract(doc.content[:3000])
            for entity in entities:
                self.kg.add_entity(entity)
            for rel in relationships:
                self.kg.add_relationship(rel)
            print(f"  엔티티 추출: {len(entities)}개, 관계: {len(relationships)}개")

        return chunks

    def search(self, query: str, n_results: int = 5, department: str = None) -> list[dict]:
        """쿼리를 검색합니다."""
        query_embedding = self.embedding.embed(query)
        where_filter = {"department": department} if department else None
        return self.vector_store.search(query_embedding, n_results, where_filter)

def main():
    parser = argparse.ArgumentParser(description="RAG 데이터 수집 파이프라인")
    parser.add_argument("--input-dir", required=True, help="문서 디렉토리 경로")
    parser.add_argument("--persist-dir", default="./chroma_db", help="벡터 DB 저장 경로")
    parser.add_argument("--no-contextual", action="store_true", help="Contextual Retrieval 비활성화")
    parser.add_argument("--build-kg", action="store_true", help="지식 그래프 구축")
    parser.add_argument("--embedding-model", default="nomic-embed-text", help="임베딩 모델")
    parser.add_argument("--llm-model", default="llama3.2", help="LLM 모델 (문맥 생성용)")
    args = parser.parse_args()

    pipeline = RAGPipeline(
        persist_dir=args.persist_dir,
        use_contextual=not args.no_contextual,
        build_knowledge_graph=args.build_kg,
        embedding_model=args.embedding_model,
        llm_model=args.llm_model,
    )

    pipeline.process_directory(args.input_dir)

if __name__ == "__main__":
    main()

이 파이프라인을 실행하는 명령은 단순합니다.

복사
# 기본 실행 (Contextual Retrieval 포함)
python pipeline.py --input-dir ./documents --persist-dir ./chroma_db

# Contextual Retrieval 없이 빠르게 인덱싱
python pipeline.py --input-dir ./documents --no-contextual

# 지식 그래프까지 구축
python pipeline.py --input-dir ./documents --build-kg

실행 전에 Ollama가 설치되어 있어야 하고, 필요한 모델을 미리 받아야 합니다.

복사
# Ollama 설치 (macOS)
brew install ollama

# 임베딩 모델 다운로드
ollama pull nomic-embed-text

# LLM 모델 다운로드 (Contextual Retrieval용)
ollama pull llama3.2

# 필요한 Python 패키지 설치
pip install chromadb langchain langchain-community pymupdf beautifulsoup4 requests

이 구성의 월간 비용은 Ollama를 로컬에서 실행하므로 사실상 무료입니다. 기존 서버나 개발 장비에서 실행할 수 있고, GPU가 없어도 CPU 모드로 동작합니다. GPU가 있으면 임베딩과 문맥 생성이 빨라지지만, 인덱싱은 배치 작업이라 속도가 크게 중요하지는 않습니다.

실무에서 반드시 고려해야 할 것이 증분 업데이트입니다. 위 파이프라인은 디렉토리 전체를 한 번에 처리하는 구조인데, 매일 새 문서가 추가되거나 기존 문서가 수정되는 환경에서 전체 재인덱싱은 비효율적입니다. 해결 방법은 간단합니다. 각 문서의 파일 경로와 해시를 기록해두고, 다음 실행 시 변경된 문서만 감지하여 해당 문서의 기존 청크를 삭제한 뒤 새로 인덱싱하면 됩니다. ChromaDB에서는 collection.delete(where={"file_path": "변경된_파일_경로"})로 특정 문서의 청크를 일괄 삭제할 수 있고, pgvector에서는 DELETE FROM document_chunks WHERE metadata->>'file_path' = '...'로 같은 작업을 수행합니다. 이 증분 처리 로직은 process_directory 메서드에 파일 해시 비교 단계를 추가하는 것만으로 구현할 수 있습니다.

마무리

이번 편에서 다룬 내용을 정리합니다.

문서 전처리에서 시작했습니다. PDF, 워드, HTML 등 여러 형식의 문서를 일관된 텍스트로 변환하고, 메타데이터를 붙이고, 중복을 걸러냅니다. 전체 프로젝트 비용의 30~50%를 차지하는 단계인데, 여기서 품질이 무너지면 그 뒤는 아무리 잘해도 소용이 없습니다.

청킹은 Recursive Character Splitting을 기본값으로 권장했습니다. 400~512 토큰에 50~100 토큰 오버랩이 출발점이고요. 구조화된 문서에는 계층적 청킹이, 의료나 법률 같은 전문 도메인에서는 시맨틱 청킹이 맞습니다.

Anthropic의 Contextual Retrieval은 각 청크에 문맥 설명을 덧붙여서 검색 실패율을 최대 67%까지 줄여줍니다. Ollama 로컬 모델을 쓰면 추가 비용도 없고요.

임베딩 모델은 Ollama의 nomic-embed-text가 출발점입니다. 상용 API보다 성능이 좋으면서 무료거든요. 벡터 저장소는 ChromaDB나 pgvector 중 환경에 맞는 걸 고르면 됩니다.

지식 그래프는 선택 사항입니다. 관계형 질문이 잦은 환경에서만 도입을 고려하면 됩니다.

이 모든 걸 하나의 스크립트로 조립했습니다. python pipeline.py --input-dir ./documents 한 줄이면 문서 로딩부터 벡터 인덱싱까지 돌아갑니다.

바로 실행해보고 싶다면, Ollama를 설치하고 nomic-embed-text와 llama3.2 모델을 받은 뒤 사내 문서 폴더를 --input-dir로 지정하면 됩니다. 문서 100개 기준, CPU 환경에서 10~20분이면 인덱싱이 끝납니다. 그 뒤로는 pipeline.search("궁금한 내용")으로 자연어 검색이 됩니다. 처음에는 --no-contextual 옵션으로 빠르게 인덱싱해보고, 검색 품질을 확인한 뒤 Contextual Retrieval을 켜는 것도 괜찮은 순서입니다.

다음 편에서는 이 데이터 위에서 실제로 검색하고 답변을 생성하는 계층을 다룹니다. 하이브리드 검색(벡터 + BM25), 리랭킹, 가드레일 시스템, 그리고 RAGAS를 이용한 품질 평가 루프까지. 데이터가 제대로 준비되어 있으면, 검색 계층의 최적화가 체감 품질을 확 끌어올립니다.

---

참고 자료

• Best Chunking Strategies for RAG - Firecrawl
• Mastering Chunking Strategies for RAG Applications - Databricks
• NAACL 2025 Findings - Semantic Chunking 비교 연구
• Anthropic Contextual Retrieval - Anthropic 공식 문서
• Ollama Embeddings - Ollama 공식 문서
• Best Embedding Models 2026 - Elephas
• LlamaIndex Production RAG - LlamaIndex 공식 문서
• RAG Implementation Cost and ROI Analysis - Stratagem Systems
• LightRAG - GitHub (HKUDS)
• Swappable LLM Backend Pattern - GitHub (nshkrdotcom)

---

다음 편: [3편] 검색, 평가, 그리고 신뢰할 수 있는 시스템