RAG 청킹 전략 심화: 검색 품질을 결정하는 문서 분할 방법

이 글은 누구를 위한 것인가

• RAG를 구축했는데 검색이 이상한 부분을 자꾸 가져오는 팀
• 청크 크기를 얼마로 해야 할지 모르는 개발자
• PDF 표, 코드 블록이 제대로 검색되지 않는 문제를 겪는 팀

---

들어가며

RAG의 90%는 청킹이 결정한다. 청크가 너무 작으면 컨텍스트가 없고, 너무 크면 관련 없는 내용이 들어온다. 문서 유형마다 최적 전략이 다르다.

이 글은 bluefoxdev.kr의 RAG 시스템 구축 가이드 를 참고하여 작성했습니다.

---

1. 청킹 전략 비교

[청킹 방법 비교]

고정 크기 청킹 (Fixed-size):
  단순: 500 토큰마다 분할
  오버랩: 50-100 토큰 (컨텍스트 연속성)
  문제: 문장 중간에서 잘림
  적합: 균질한 산문 텍스트

재귀 분할 (Recursive):
  구분자: \n\n → \n → 문장 → 단어 순
  문단 → 문장 경계에서 분할
  적합: 대부분의 텍스트 문서

시맨틱 청킹 (Semantic):
  임베딩 유사도로 의미 경계 감지
  내용이 바뀌는 지점에서 분할
  품질: 높음, 비용: 높음
  적합: 의미 단위가 중요한 문서

부모-자식 청킹 (Parent-Child):
  작은 청크로 검색 → 큰 청크 반환
  정밀한 검색 + 충분한 컨텍스트
  적합: 긴 기술 문서

[문서 유형별 전략]
  일반 문서: 재귀 분할, 512-1024 토큰
  코드: 함수/클래스 단위 분할
  테이블: 행 또는 테이블 전체 유지
  PDF: 페이지/섹션 헤더 기반
  대화 기록: 메시지 단위

---

2. 고급 청킹 구현

import re
import anthropic
from dataclasses import dataclass, field

client = anthropic.Anthropic()

@dataclass
class Chunk:
    text: str
    metadata: dict = field(default_factory=dict)
    chunk_id: str = ""
    parent_id: str | None = None
    start_char: int = 0
    end_char: int = 0

def recursive_split(
    text: str,
    chunk_size: int = 1000,
    chunk_overlap: int = 100,
    separators: list[str] | None = None,
) -> list[str]:
    """재귀적 텍스트 분할"""
    
    if separators is None:
        separators = ["\n\n", "\n", "。", ".", " "]
    
    if len(text) <= chunk_size:
        return [text]
    
    for separator in separators:
        if separator in text:
            parts = text.split(separator)
            
            chunks = []
            current = ""
            
            for part in parts:
                if len(current) + len(part) + len(separator) <= chunk_size:
                    current += (separator if current else "") + part
                else:
                    if current:
                        chunks.append(current)
                    current = part
            
            if current:
                chunks.append(current)
            
            # 오버랩 추가
            result = [chunks[0]]
            for i in range(1, len(chunks)):
                overlap_text = result[-1][-chunk_overlap:]
                result.append(overlap_text + chunks[i])
            
            return result
    
    # 마지막 수단: 문자 기준 분할
    return [text[i:i+chunk_size] for i in range(0, len(text), chunk_size - chunk_overlap)]

def parent_child_chunks(
    text: str,
    parent_size: int = 2000,
    child_size: int = 400,
) -> tuple[list[Chunk], list[Chunk]]:
    """부모-자식 계층 청킹"""
    import uuid
    
    parent_texts = recursive_split(text, chunk_size=parent_size, chunk_overlap=50)
    parents = []
    children = []
    
    for i, parent_text in enumerate(parent_texts):
        parent_id = str(uuid.uuid4())[:8]
        parent = Chunk(
            text=parent_text,
            chunk_id=parent_id,
            metadata={"type": "parent", "index": i},
        )
        parents.append(parent)
        
        child_texts = recursive_split(parent_text, chunk_size=child_size, chunk_overlap=50)
        for j, child_text in enumerate(child_texts):
            child = Chunk(
                text=child_text,
                chunk_id=f"{parent_id}-{j}",
                parent_id=parent_id,
                metadata={"type": "child", "parent_index": i, "child_index": j},
            )
            children.append(child)
    
    return parents, children

def split_code_by_function(code: str, language: str = "python") -> list[Chunk]:
    """코드를 함수/클래스 단위로 분할"""
    
    chunks = []
    
    if language == "python":
        # 함수와 클래스 정의 경계에서 분할
        pattern = r"(?=^(?:def |class |async def )\s*\w+)"
        parts = re.split(pattern, code, flags=re.MULTILINE)
        
        for i, part in enumerate(parts):
            if part.strip():
                chunk = Chunk(
                    text=part,
                    metadata={"language": language, "type": "code_block", "index": i},
                )
                chunks.append(chunk)
    
    return chunks

async def semantic_chunking(
    text: str,
    similarity_threshold: float = 0.85,
) -> list[Chunk]:
    """시맨틱 청킹: 의미 경계에서 분할"""
    
    # 문장 분할
    sentences = re.split(r"(?<=[.!?。])\s+", text)
    
    if len(sentences) < 3:
        return [Chunk(text=text)]
    
    # 인접 문장 유사도 계산
    import openai, numpy as np
    oai = openai.OpenAI()
    
    embeddings_resp = oai.embeddings.create(
        input=sentences,
        model="text-embedding-3-small",
    )
    embeddings = [e.embedding for e in embeddings_resp.data]
    
    # 유사도 낮은 지점에서 청크 경계
    chunks = []
    current_sentences = [sentences[0]]
    
    for i in range(1, len(sentences)):
        prev_emb = np.array(embeddings[i-1])
        curr_emb = np.array(embeddings[i])
        similarity = np.dot(prev_emb, curr_emb) / (
            np.linalg.norm(prev_emb) * np.linalg.norm(curr_emb)
        )
        
        if similarity < similarity_threshold:
            chunks.append(Chunk(
                text=" ".join(current_sentences),
                metadata={"type": "semantic"},
            ))
            current_sentences = [sentences[i]]
        else:
            current_sentences.append(sentences[i])
    
    if current_sentences:
        chunks.append(Chunk(
            text=" ".join(current_sentences),
            metadata={"type": "semantic"},
        ))
    
    return chunks

def enrich_chunk_metadata(
    chunk: Chunk,
    document_metadata: dict,
    chunk_index: int,
    total_chunks: int,
) -> Chunk:
    """청크 메타데이터 풍부화"""
    chunk.metadata.update({
        **document_metadata,
        "chunk_index": chunk_index,
        "total_chunks": total_chunks,
        "position": "start" if chunk_index == 0 else "end" if chunk_index == total_chunks - 1 else "middle",
        "char_count": len(chunk.text),
    })
    return chunk

---

마무리

청킹 전략 선택: 일반 문서는 재귀 분할 500-1000 토큰, 긴 기술 문서는 부모-자식 계층, 코드베이스는 함수 단위. 청크 크기 최적화는 5가지 크기(256, 512, 1024, 2048, 4096)로 A/B 테스트하면 2-3시간 만에 최적값을 찾을 수 있다. 메타데이터(섹션, 출처, 날짜)를 청크에 풍부하게 담으면 필터링 검색도 가능해진다.