LLM RAG 파이프라인: 청킹 전략과 임베딩 최적화 실전 2026

from langchain.text_splitter import CharacterTextSplitter

# 고정 크기 청킹 - 가장 기본적인 방식
splitter = CharacterTextSplitter(
    separator="\n",
    chunk_size=512,
    chunk_overlap=50,       # 10% 오버랩으로 문맥 유지
    length_function=len,
)

documents = splitter.split_text(raw_text)
print(f"총 청크 수: {len(documents)}")
print(f"평균 청크 길이: {sum(len(d) for d in documents) / len(documents):.0f}자")

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 2026년 벤치마크 기준 최적 설정
splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,          # 약 12% 오버랩
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len,
    is_separator_regex=False,
)

chunks = splitter.split_text(raw_text)

# 청크 품질 검증
for i, chunk in enumerate(chunks[:3]):
    print(f"[Chunk {i}] 길이={len(chunk)} | 시작: {chunk[:80]}...")

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# 시맨틱 청킹 - 의미 전환점 기반 분할
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
semantic_splitter = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",  # percentile, standard_deviation, interquartile
    breakpoint_threshold_amount=75,          # 상위 25% 유사도 차이에서 분할
)

semantic_chunks = semantic_splitter.split_text(raw_text)
print(f"시맨틱 청크 수: {len(semantic_chunks)}")
print(f"평균 길이: {sum(len(c) for c in semantic_chunks) / len(semantic_chunks):.0f}자")

from langchain.text_splitter import MarkdownHeaderTextSplitter

# Markdown 구조 기반 청킹
headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on,
    strip_headers=False,
)

md_chunks = md_splitter.split_text(markdown_text)

# 각 청크에 메타데이터로 헤더 계층 정보 포함
for chunk in md_chunks[:3]:
    print(f"메타데이터: {chunk.metadata}")
    print(f"내용: {chunk.page_content[:100]}...")
    print("---")

from openai import OpenAI
import numpy as np

client = OpenAI()

def generate_embeddings(
    texts: list[str],
    model: str = "text-embedding-3-large",
    dimensions: int = 1024,    # 차원 축소로 비용/속도 최적화
    batch_size: int = 100,
) -> np.ndarray:
    """배치 단위 임베딩 생성 with 차원 축소"""
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = client.embeddings.create(
            input=batch,
            model=model,
            dimensions=dimensions,  # text-embedding-3 시리즈만 지원
        )
        batch_embs = [item.embedding for item in response.data]
        all_embeddings.extend(batch_embs)

    return np.array(all_embeddings, dtype=np.float32)


# 사용 예시
chunks = ["RAG 파이프라인의 핵심은 검색 품질이다.", "청킹 전략에 따라 결과가 달라진다."]
embeddings = generate_embeddings(chunks, dimensions=1024)
print(f"임베딩 shape: {embeddings.shape}")  # (2, 1024)

from FlagEmbedding import BGEM3FlagModel

# BGE-M3: dense + sparse + colbert 동시 지원
model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)

sentences = [
    "LLM RAG 파이프라인에서 청킹 전략은 검색 품질의 핵심이다.",
    "벡터 데이터베이스 인덱싱은 검색 지연시간에 직접 영향을 준다.",
]

# dense + sparse + colbert 임베딩 동시 생성
output = model.encode(
    sentences,
    batch_size=12,
    max_length=512,
    return_dense=True,
    return_sparse=True,
    return_colbert_vecs=True,
)

dense_embeddings = output["dense_vecs"]       # shape: (2, 1024)
sparse_embeddings = output["lexical_weights"]  # 희소 벡터 (BM25 대체)
colbert_vecs = output["colbert_vecs"]          # multi-vector (정밀 매칭)

print(f"Dense shape: {dense_embeddings.shape}")
print(f"Sparse keys 예시: {list(sparse_embeddings[0].keys())[:5]}")

import chromadb
from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction

# Chroma 클라이언트 초기화 (영구 저장)
client = chromadb.PersistentClient(path="./chroma_db")

embedding_fn = OpenAIEmbeddingFunction(
    api_key="sk-...",
    model_name="text-embedding-3-large",
)

# 컬렉션 생성 (HNSW 인덱스 자동 적용)
collection = client.get_or_create_collection(
    name="rag_knowledge_base",
    embedding_function=embedding_fn,
    metadata={
        "hnsw:space": "cosine",       # 유사도 메트릭
        "hnsw:M": 32,                 # HNSW 연결 수 (높을수록 정확, 메모리 증가)
        "hnsw:ef_construction": 200,  # 인덱스 구축 시 탐색 폭
    },
)

# 문서 추가 (배치)
collection.add(
    documents=["RAG에서 청킹은 검색 품질의 80%를 결정한다.", "임베딩 모델 선택이 나머지 20%를 좌우한다."],
    metadatas=[
        {"source": "rag_guide", "section": "chunking", "date": "2026-03"},
        {"source": "rag_guide", "section": "embedding", "date": "2026-03"},
    ],
    ids=["doc_001", "doc_002"],
)

# 검색 (메타데이터 필터 + 유사도)
results = collection.query(
    query_texts=["RAG 파이프라인에서 가장 중요한 요소는?"],
    n_results=5,
    where={"source": "rag_guide"},
    include=["documents", "distances", "metadatas"],
)

for doc, dist, meta in zip(
    results["documents"][0], results["distances"][0], results["metadatas"][0]
):
    print(f"[거리: {dist:.4f}] {meta['section']} | {doc[:80]}")

import numpy as np
from typing import List, Set


def mean_reciprocal_rank(
    retrieved_ids: List[List[str]],
    relevant_ids: List[Set[str]],
) -> float:
    """MRR 계산: 각 쿼리에서 첫 번째 관련 문서의 순위 역수 평균"""
    mrr_scores = []
    for retrieved, relevant in zip(retrieved_ids, relevant_ids):
        for rank, doc_id in enumerate(retrieved, 1):
            if doc_id in relevant:
                mrr_scores.append(1.0 / rank)
                break
        else:
            mrr_scores.append(0.0)
    return np.mean(mrr_scores)


def recall_at_k(
    retrieved_ids: List[List[str]],
    relevant_ids: List[Set[str]],
    k: int = 10,
) -> float:
    """Recall@K: 상위 K개 결과에서 관련 문서 비율"""
    recalls = []
    for retrieved, relevant in zip(retrieved_ids, relevant_ids):
        top_k = set(retrieved[:k])
        if len(relevant) == 0:
            continue
        recalls.append(len(top_k & relevant) / len(relevant))
    return np.mean(recalls)


def ndcg_at_k(
    retrieved_ids: List[List[str]],
    relevant_ids: List[Set[str]],
    k: int = 10,
) -> float:
    """NDCG@K: 순위 가중 관련성 평가"""
    ndcg_scores = []
    for retrieved, relevant in zip(retrieved_ids, relevant_ids):
        # DCG 계산
        dcg = 0.0
        for rank, doc_id in enumerate(retrieved[:k], 1):
            if doc_id in relevant:
                dcg += 1.0 / np.log2(rank + 1)

        # Ideal DCG 계산
        ideal_hits = min(len(relevant), k)
        idcg = sum(1.0 / np.log2(r + 1) for r in range(1, ideal_hits + 1))

        ndcg_scores.append(dcg / idcg if idcg > 0 else 0.0)
    return np.mean(ndcg_scores)


# 사용 예시
retrieved = [["d1", "d3", "d5", "d2", "d4"]]
relevant = [{"d1", "d2", "d7"}]

print(f"MRR:       {mean_reciprocal_rank(retrieved, relevant):.4f}")
print(f"Recall@3:  {recall_at_k(retrieved, relevant, k=3):.4f}")
print(f"Recall@5:  {recall_at_k(retrieved, relevant, k=5):.4f}")
print(f"NDCG@5:    {ndcg_at_k(retrieved, relevant, k=5):.4f}")

from qdrant_client import QdrantClient, models
from qdrant_client.models import Distance, VectorParams, SparseVectorParams

client = QdrantClient(host="localhost", port=6333)

# Dense + Sparse 벡터를 동시에 저장하는 컬렉션 생성
client.create_collection(
    collection_name="hybrid_rag",
    vectors_config={
        "dense": VectorParams(size=1024, distance=Distance.COSINE),
    },
    sparse_vectors_config={
        "sparse": SparseVectorParams(),
    },
)

# 문서 색인 (dense + sparse 벡터 동시 저장)
client.upsert(
    collection_name="hybrid_rag",
    points=[
        models.PointStruct(
            id=1,
            vector={
                "dense": dense_embedding.tolist(),
                "sparse": models.SparseVector(
                    indices=list(sparse_weights.keys()),
                    values=list(sparse_weights.values()),
                ),
            },
            payload={"text": "RAG 파이프라인 청킹 가이드", "source": "blog"},
        ),
    ],
)

# Hybrid Search 실행 (RRF 기반 점수 병합)
results = client.query_points(
    collection_name="hybrid_rag",
    prefetch=[
        models.Prefetch(
            query=dense_query_vector.tolist(),
            using="dense",
            limit=20,
        ),
        models.Prefetch(
            query=models.SparseVector(
                indices=list(sparse_query.keys()),
                values=list(sparse_query.values()),
            ),
            using="sparse",
            limit=20,
        ),
    ],
    query=models.FusionQuery(fusion=models.Fusion.RRF),  # Reciprocal Rank Fusion
    limit=10,
)

for point in results.points:
    print(f"[Score: {point.score:.4f}] {point.payload['text']}")