rag-구현

RAG 구현 패턴

Retrieval-Augmented Generation (RAG) 시스템 구현에 대한 포괄적인 가이드. 벡터 데이터베이스 선택, 청킹 전략, 임베딩 모델, 검색 최적화 및 프로덕션 배포 패턴을 다룹니다.

---

빠른 참조

이 스킬을 사용해야 하는 경우:

• RAG/시맨틱 검색 시스템 구축
• 문서 검색 파이프라인 구현
• 벡터 데이터베이스 성능 최적화
• 검색 품질 문제 디버깅
• 벡터 데이터베이스 옵션 선택
• 청킹 전략 설계
• 하이브리드 검색 구현

다루는 기술:

• 벡터 DB: Qdrant, Pinecone, Chroma, Weaviate, Milvus
• 임베딩: OpenAI, Sentence Transformers, Cohere
• 프레임워크: LangChain, LlamaIndex, Haystack

---

1부: 벡터 데이터베이스 선택

데이터베이스 비교 매트릭스

데이터베이스	최적 사용 사례	배포 방식	성능	비용
Qdrant	자체 호스팅, 프로덕션	Docker/K8s	우수 (Rust)	무료 (자체 호스팅)
Pinecone	관리형, 빠른 프로토타이핑	클라우드	우수	사용량 기반 과금
Chroma	로컬 개발, 임베디드	인프로세스	좋음 (Python)	무료
Weaviate	복잡한 스키마, GraphQL	Docker/클라우드	우수 (Go)	무료 + 클라우드
Milvus	대규모 분산 환경	K8s	우수 (C++)	무료 (자체 호스팅)

Qdrant 설정 (프로덕션 권장)

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

# 클라이언트 초기화 (로컬 또는 클라우드)
client = QdrantClient(url="http://localhost:6333")  # 또는 클라우드 URL

# 컬렉션 생성
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,  # OpenAI text-embedding-3-small 차원
        distance=Distance.COSINE  # 또는 DOT, EUCLID
    )
)

# 페이로드와 함께 벡터 삽입
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=[0.1, 0.2, ...],  # 1536 차원
            payload={
                "text": "문서 내용",
                "source": "doc.pdf",
                "page": 1,
                "metadata": {...}
            }
        )
    ]
)

# 검색
results = client.search(
    collection_name="documents",
    query_vector=[0.1, 0.2, ...],
    limit=5,
    score_threshold=0.7  # 최소 유사도
)

Pinecone 설정 (관리형 서비스)

from pinecone import Pinecone, ServerlessSpec

# 초기화
pc = Pinecone(api_key="your-key")

# 인덱스 생성
pc.create_index(
    name="documents",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1")
)

# 인덱스 가져오기
index = pc.Index("documents")

# 벡터 업서트
index.upsert(vectors=[
    ("doc1", [0.1, 0.2, ...], {"text": "...", "source": "..."})
])

# 쿼리
results = index.query(
    vector=[0.1, 0.2, ...],
    top_k=5,
    include_metadata=True
)

---

2부: 청킹 전략

전략 1: 고정 크기 청킹 (간단, 빠름)

def fixed_size_chunking(text: str, chunk_size: int = 512, overlap: int = 50) -> list[str]:
    """
    텍스트를 중첩(overlap)이 있는 고정 크기 청크로 분할합니다.

    장점: 간단하고 예측 가능한 청크 크기
    단점: 문장 중간에 끊길 수 있으며 의미적 경계가 불량함
    """
    words = text.split()
    chunks = []

    for i in range(0, len(words), chunk_size - overlap):
        chunk = ' '.join(words[i:i + chunk_size])
        chunks.append(chunk)

    return chunks

# 사용법
chunks = fixed_size_chunking(document, chunk_size=512, overlap=50)

사용 시기:

• 간단한 문서 (로그, 대본)
• 프로토타이핑/MVP
• 일관된 토큰 예산 필요 시

전략 2: 의미적 청킹 (더 나은 품질)

from langchain.text_splitter import RecursiveCharacterTextSplitter

def semantic_chunking(text: str, chunk_size: int = 1000, overlap: int = 200) -> list[str]:
    """
    의미적 경계 (문단, 문장)를 기준으로 분할합니다.

    장점: 의미 보존, 더 나은 검색 품질
    단점: 가변 청크 크기, 느린 처리 속도
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", ". ", " ", ""],  # 우선 순위 순서
        length_function=len
    )

    return splitter.split_text(text)

# 사용법
chunks = semantic_chunking(document, chunk_size=1000, overlap=200)

사용 시기:

• 긴 형식의 문서 (기사, 책, 보고서)
• 품질 > 속도
• 자연어 콘텐츠

전략 3: 계층적 청킹 (구조화된 문서에 최적)

def hierarchical_chunking(document: dict) -> list[dict]:
    """
    문서 구조 (섹션, 하위 섹션)를 기반으로 청킹합니다.

    장점: 계층 구조 보존, 부모-자식 검색 가능
    단점: 구조화된 입력 필요, 더 복잡함
    """
    chunks = []

    for section in document['sections']:
        # 부모 청크 (섹션 요약)
        chunks.append({
            'text': section['title'] + '\n' + section['summary'],
            'type': 'parent',
            'section_id': section['id']
        })

        # 자식 청크 (문단)
        for para in section['paragraphs']:
            chunks.append({
                'text': para,
                'type': 'child',
                'parent_id': section['id']
            })

    return chunks

사용 시기:

• 기술 문서
• 목차가 있는 책
• 법률 문서
• 컨텍스트 계층 구조 보존 필요 시

전략 4: 슬라이딩 윈도우 (최대 컨텍스트 보존)

def sliding_window_chunking(text: str, window_size: int = 512, stride: int = 256) -> list[str]:
    """
    최대 컨텍스트를 위한 중첩 윈도우입니다.

    장점: 경계에서 정보 손실 없음
    단점: 저장 공간 오버헤드 (중복 콘텐츠)
    """
    words = text.split()
    chunks = []

    for i in range(0, len(words) - window_size + 1, stride):
        chunk = ' '.join(words[i:i + window_size])
        chunks.append(chunk)

    return chunks

사용 시기:

• 중요한 검색 정확도 필요 시
• 짧은 쿼리가 더 넓은 컨텍스트 필요 시
• 저장 비용이 문제되지 않을 때

---

3부: 임베딩 모델

모델 선택 가이드

모델	차원	속도	품질	비용	사용 사례
OpenAI text-embedding-3-small	1536	빠름	우수	$0.02/1M 토큰	프로덕션, 일반 목적
OpenAI text-embedding-3-large	3072	중간	최고	$0.13/1M 토큰	고품질 검색
all-MiniLM-L6-v2	384	매우 빠름	좋음	무료	자체 호스팅, 프로토타이핑
all-mpnet-base-v2	768	빠름	매우 좋음	무료	자체 호스팅, 품질
Cohere embed-english-v3.0	1024	빠름	우수	$0.10/1M 토큰	시맨틱 검색 중심

OpenAI 임베딩 (권장)

from openai import OpenAI

client = OpenAI(api_key="your-key")

def get_embeddings(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    """
    OpenAI를 사용하여 임베딩을 생성합니다.

    배치 크기: 요청당 최대 2048개 입력
    속도 제한: 티어 제한 확인
    """
    response = client.embeddings.create(
        model=model,
        input=texts
    )

    return [item.embedding for item in response.data]

# 사용법
chunks = ["청크 1", "청크 2", ...]
embeddings = get_embeddings(chunks)

Sentence Transformers (자체 호스팅)

from sentence_transformers import SentenceTransformer

# 모델 로드 (첫 다운로드 후 캐시됨)
model = SentenceTransformer('all-MiniLM-L6-v2')

def get_embeddings_local(texts: list[str]) -> list[list[float]]:
    """
    로컬에서 임베딩을 생성합니다 (API 비용 없음).

    100개 이상의 배치에는 GPU 권장
    소규모 배치에는 CPU 사용 가능
    """
    return model.encode(texts, show_progress_bar=True).tolist()

# 사용법
embeddings = get_embeddings_local(chunks)

---

4부: 검색 최적화

기술 1: 하이브리드 검색 (밀집 + 희소)

from qdrant_client.models import Filter, FieldCondition, MatchValue

def hybrid_search(query: str, query_vector: list[float], top_k: int = 10):
    """
    밀집(벡터) 검색과 희소(키워드) 검색을 결합합니다.

    밀집: 시맨틱 유사도
    희소: 정확한 키워드 일치
    """
    # 밀집 검색
    dense_results = client.search(
        collection_name="documents",
        query_vector=query_vector,
        limit=top_k * 2  # 더 많은 후보 확보
    )

    # 희소 검색 (메타데이터를 통한 BM25)
    sparse_results = client.search(
        collection_name="documents",
        query_filter=Filter(
            must=[
                FieldCondition(
                    key="text",
                    match=MatchValue(value=query)
                )
            ]
        ),
        limit=top_k * 2
    )

    # 병합 및 재순위화
    combined = merge_results(dense_results, sparse_results, weights=(0.7, 0.3))
    return combined[:top_k]

기술 2: 쿼리 확장

def expand_query(query: str) -> list[str]:
    """
    더 나은 재현율을 위해 쿼리 변형을 생성합니다.

    기술:
    - 동의어 확장
    - 질문 재구성
    - 엔티티 추출
    """
    from openai import OpenAI
    client = OpenAI()

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "사용자 쿼리의 대체 표현 3개를 생성하세요."
        }, {
            "role": "user",
            "content": query
        }]
    )

    expanded = [query] + response.choices[0].message.content.split('\n')
    return expanded

# 사용법
queries = expand_query("신경망을 학습하는 방법은?")
# → ["신경망을 학습하는 방법은?",
#    "신경망 학습 기술은 무엇인가?",
#    "신경망 최적화 방법",
#    "딥러닝 모델 학습"]

기술 3: 재순위화

from sentence_transformers import CrossEncoder

# 크로스 인코더 로드 (재순위화를 위해 바이 인코더보다 우수)
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def rerank_results(query: str, results: list[dict], top_k: int = 5) -> list[dict]:
    """
    크로스 인코더를 사용하여 초기 결과를 재순위화합니다.

    초기 검색보다 정확하지만 느림
    상위 20-50개 후보에만 사용
    """
    # 각 쿼리-문서 쌍 점수 매기기
    pairs = [(query, result['text']) for result in results]
    scores = reranker.predict(pairs)

    # 점수를 결과와 결합
    for result, score in zip(results, scores):
        result['rerank_score'] = float(score)

    # 정렬 후 상위 top_k 반환
    reranked = sorted(results, key=lambda x: x['rerank_score'], reverse=True)
    return reranked[:top_k]

기술 4: 메타데이터 필터링

def filtered_search(
    query_vector: list[float],
    filters: dict,
    top_k: int = 5
):
    """
    메타데이터(날짜, 카테고리, 작성자 등)로 검색을 필터링합니다.

    사전 필터: 더 빠르지만 결과를 놓칠 수 있음
    사후 필터: 더 많은 결과를 제공하지만 느림
    """
    from qdrant_client.models import Filter, FieldCondition, Range

    # 필터 조건 생성
    conditions = []

    if 'date_range' in filters:
        conditions.append(
            FieldCondition(
                key="date",
                range=Range(
                    gte=filters['date_range']['start'],
                    lte=filters['date_range']['end']
                )
            )
        )

    if 'category' in filters:
        conditions.append(
            FieldCondition(
                key="category",
                match=MatchValue(value=filters['category'])
            )
        )

    # 필터로 검색
    results = client.search(
        collection_name="documents",
        query_vector=query_vector,
        query_filter=Filter(must=conditions) if conditions else None,
        limit=top_k
    )

    return results

---

5부: 컨텍스트 관리

패턴 1: 검색된 컨텍스트 최적화

def optimize_context(query: str, retrieved_docs: list[str], max_tokens: int = 4000) -> str:
    """
    검색된 컨텍스트를 LLM 컨텍스트 창에 맞게 최적화합니다.

    전략:
    1. 관련성 기반 자르기
    2. 추출 요약
    3. 중복 제거
    """
    # 관련성별 정렬
    sorted_docs = sorted(retrieved_docs, key=lambda d: d['score'], reverse=True)

    # 토큰 예산 내에서 컨텍스트 구축
    context_parts = []
    total_tokens = 0

    for doc in sorted_docs:
        doc_tokens = estimate_tokens(doc['text'])

        if total_tokens + doc_tokens <= max_tokens:
            context_parts.append(f"[출처: {doc['source']}]\n{doc['text']}")
            total_tokens += doc_tokens
        else:
            # 마지막 문서를 맞게 자르기
            remaining = max_tokens - total_tokens
            truncated = truncate_to_tokens(doc['text'], remaining)
            context_parts.append(f"[출처: {doc['source']}]\n{truncated}")
            break

    return "\n\n---\n\n".join(context_parts)

패턴 2: 인용 추적

def generate_with_citations(query: str, context: str, sources: list[dict]) -> dict:
    """
    인용 추적과 함께 답변을 생성합니다.

    반환:
    - answer: 생성된 텍스트
    - citations: 사용된 출처 문서 목록
    """
    from openai import OpenAI
    client = OpenAI()

    # 출처 맵 생성
    source_map = {i+1: source for i, source in enumerate(sources)}
    numbered_context = "\n\n".join([
        f"[{i+1}] {source['text']}"
        for i, source in enumerate(sources)
    ])

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "제공된 출처를 사용하여 답변하세요. 출처는 [1], [2] 등으로 인용하세요."
        }, {
            "role": "user",
            "content": f"컨텍스트:\n{numbered_context}\n\n질문: {query}"
        }]
    )

    answer = response.choices[0].message.content

    return {
        "answer": answer,
        "citations": extract_citations(answer, source_map)
    }

패턴 3: 대화형 검색 (멀티턴 RAG)

class ConversationRAG:
    """
    대화 컨텍스트로 검색을 향상시키는 멀티턴 RAG 시스템입니다.
    """
    def __init__(self, vector_store, llm_client):
        self.vector_store = vector_store
        self.llm = llm_client
        self.history = []

    def query(self, user_input: str, top_k: int = 5) -> str:
        # 히스토리로 쿼리 재구성
        query = self._rewrite_with_context(user_input)

        # 관련 문서 검색
        results = self.vector_store.search(query, top_k=top_k)

        # 컨텍스트와 히스토리로 답변 생성
        answer = self._generate_answer(query, results, self.history)

        # 히스토리 업데이트
        self.history.append({"user": user_input, "assistant": answer})

        return answer

---

6부: 프로덕션 배포

시스템 아키텍처

┌──────────┐     ┌──────────┐     ┌──────────┐
│  문서 처리  │────▶│  벡터 저장소  │────▶│  검색 엔진   │
└──────────┘     └──────────┘     └──────────┘
       │                              │
       ▼                              ▼
┌──────────┐                    ┌──────────┐
│  청킹 파이프라인│                    │  쿼리 최적화  │
└──────────┘                    └──────────┘
       │                              │
       ▼                              ▼
┌──────────┐                    ┌──────────┐
│  임베딩 모델 │                    │  LLM  추론 │
└──────────┘                    └──────────┘
       │                              │
       └──────────┬──────────┬──────────┘
                  ▼          ▼
            ┌──────────┐ ┌──────────┐
            │  캐싱 계층  │ │  모니터링   │
            └──────────┘ └──────────┘

성능 최적화 팁

1. 벡터 인덱싱: 대규모 데이터셋의 경우 IVF (Inverted File) 또는 HNSW (Hierarchical Navigable Small World) 인덱스 사용
2. 캐싱: 자주 액세스하는 쿼리 결과를 Redis에 캐싱
3. 일괄 처리: 임베딩 요청을 배치로 묶어 API 호출 최소화
4. 비동기: 비동기 클라이언트를 위해 asyncio 사용
5. 압축: 차원 축소를 위해 벡터 양자화 또는 PCA 사용

모니터링 메트릭

메트릭	목표	임계값
검색 정밀도	> 0.85	< 0.70
검색 재현율	> 0.80	< 0.60
엔드투엔드 지연 시간	< 2초	> 5초
벡터 DB 가용성	99.9%	< 99.5%
토큰 사용량	예산 내	예산 초과 시 경고

---

모범 사례 요약

시작하기

• Qdrant 또는 Pinecone으로 빠른 프로토타이핑부터 시작
• 의미적 청킹과 기본 임베딩 사용
• 검색 품질을 측정하는 평가 세트 구축

프로덕션 준비

• 정밀도를 높이기 위해 하이브리드 검색 구현
• 재순위화를 위해 크로스 인코더 사용
• 지연 시간을 줄이기 위해 캐싱 추가

고급 최적화

• 검색을 안내하는 메타데이터 필터링
• 더 나은 재현율을 위해 쿼리 확장 사용
• 컨텍스트를 LLM 내부에 맞게 최적화

일반적인 함정

• ❌ 청킹 전략 무시
• ❌ 검색 품질 평가 실패
• ❌ 컨텍스트 창 제한 무시
• ❌ 피드백 루프 없음