RAG 성능 최적화 기법

RAG 시스템 최적화 개요

RAG 시스템의 성능은 다음 세 가지 측면에서 최적화할 수 있습니다:

1. 검색 품질 - 관련성 높은 문서 찾기
2. 응답 품질 - 정확하고 유용한 답변 생성
3. 성능/비용 - 빠른 응답과 비용 효율성

1. 청킹 전략 최적화

1.1 고정 크기 청킹

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 기본 설정
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
)

# 문서 유형별 최적화
## 기술 문서 - 작은 청크
tech_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=100
)

## 소설/긴 글 - 큰 청크
narrative_splitter = RecursiveCharacterTextSplitter(
    chunk_size=2000,
    chunk_overlap=400
)

1.2 의미 기반 청킹

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings

# 의미론적 유사도로 분할
semantic_chunker = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",  # or "standard_deviation"
    breakpoint_threshold_amount=95  # 상위 5%에서 분할
)

chunks = semantic_chunker.split_documents(documents)

1.3 문서 구조 기반 청킹

from langchain.text_splitter import MarkdownHeaderTextSplitter

# Markdown 헤더 기준 분할
headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

markdown_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on
)

md_chunks = markdown_splitter.split_text(markdown_document)

1.4 청크에 컨텍스트 추가

def add_context_to_chunks(chunks, context_window=1):
    """이전/다음 청크 정보를 메타데이터에 추가"""
    enhanced_chunks = []

    for i, chunk in enumerate(chunks):
        # 이전 청크의 마지막 부분
        prev_context = ""
        if i > 0:
            prev_context = chunks[i-1].page_content[-100:]

        # 다음 청크의 시작 부분
        next_context = ""
        if i < len(chunks) - 1:
            next_context = chunks[i+1].page_content[:100]

        chunk.metadata["prev_context"] = prev_context
        chunk.metadata["next_context"] = next_context
        enhanced_chunks.append(chunk)

    return enhanced_chunks

2. 임베딩 최적화

2.1 임베딩 모델 선택

from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings

# OpenAI - 고품질, 유료
openai_embeddings = OpenAIEmbeddings(
    model="text-embedding-3-large",  # 3072 차원
    # model="text-embedding-3-small",  # 1536 차원 (더 빠름, 저렴)
)

# 오픈소스 - 무료, 로컬 실행
hf_embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-large-en-v1.5",  # 영어
    # model_name="BAAI/bge-m3",  # 다국어
    model_kwargs={'device': 'cpu'},  # 또는 'cuda'
    encode_kwargs={'normalize_embeddings': True}
)

# 한국어 특화
ko_embeddings = HuggingFaceEmbeddings(
    model_name="jhgan/ko-sroberta-multitask"
)

2.2 차원 축소 (OpenAI Embeddings)

# 비용 절감을 위한 차원 축소
small_embeddings = OpenAIEmbeddings(
    model="text-embedding-3-large",
    dimensions=1024  # 기본 3072에서 축소
)

2.3 배치 임베딩

from langchain_community.vectorstores import Chroma

# 대량 문서 임베딩 - 배치 처리
def batch_embed_documents(docs, embeddings, batch_size=100):
    vectorstore = None

    for i in range(0, len(docs), batch_size):
        batch = docs[i:i+batch_size]

        if vectorstore is None:
            vectorstore = Chroma.from_documents(
                batch,
                embeddings,
                persist_directory="./chroma_db"
            )
        else:
            vectorstore.add_documents(batch)

        print(f"Processed {min(i+batch_size, len(docs))}/{len(docs)} documents")

    return vectorstore

3. 검색 최적화

3.1 하이브리드 검색 (BM25 + Vector)

from langchain.retrievers import BM25Retriever, EnsembleRetriever
from langchain_community.vectorstores import Chroma

# 키워드 검색 (BM25)
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

# 벡터 검색
vectorstore = Chroma.from_documents(documents, embeddings)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 하이브리드 (가중치 조정 가능)
hybrid_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # BM25 40%, Vector 60%
)

results = hybrid_retriever.get_relevant_documents("query")

3.2 MMR (Maximum Marginal Relevance)

# 다양성을 고려한 검색 - 중복 방지
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 5,           # 최종 반환 문서 수
        "fetch_k": 20,    # 초기 후보 문서 수
        "lambda_mult": 0.5  # 0: 최대 다양성, 1: 최대 관련성
    }
)

3.3 재순위화 (Re-ranking)

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank

# Cohere Reranker 사용
compressor = CohereRerank(
    cohere_api_key="YOUR_API_KEY",
    top_n=3  # 상위 3개만 선택
)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vector_retriever
)

# 또는 LLM 기반 재순위화
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(temperature=0)
compressor = LLMChainExtractor.from_llm(llm)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vector_retriever
)

3.4 메타데이터 필터링

# 메타데이터 기반 사전 필터링
retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 5,
        "filter": {
            "source": "technical_docs",
            "year": {"$gte": 2023}
        }
    }
)

# 또는 동적 필터링
def get_filtered_retriever(vectorstore, category=None, date_from=None):
    filter_dict = {}

    if category:
        filter_dict["category"] = category
    if date_from:
        filter_dict["date"] = {"$gte": date_from}

    return vectorstore.as_retriever(
        search_kwargs={"k": 5, "filter": filter_dict}
    )

4. 쿼리 최적화

4.1 쿼리 확장

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# 쿼리 확장 프롬프트
query_expansion_prompt = PromptTemplate(
    input_variables=["question"],
    template="""원래 질문: {question}

이 질문과 관련된 3가지 유사한 질문을 생성하세요:
1.
2.
3."""
)

expansion_chain = LLMChain(llm=llm, prompt=query_expansion_prompt)

# 원본 + 확장 쿼리로 검색
def search_with_expansion(question, retriever):
    # 원본 쿼리로 검색
    original_docs = retriever.get_relevant_documents(question)

    # 쿼리 확장
    expanded = expansion_chain.run(question=question)
    expanded_questions = expanded.split('\n')

    # 확장된 쿼리로 추가 검색
    all_docs = original_docs
    for exp_q in expanded_questions:
        exp_docs = retriever.get_relevant_documents(exp_q.strip())
        all_docs.extend(exp_docs)

    # 중복 제거
    unique_docs = list({doc.page_content: doc for doc in all_docs}.values())
    return unique_docs[:5]

4.2 HyDE (Hypothetical Document Embeddings)

from langchain.chains import HypotheticalDocumentEmbedder

# 가상 답변 생성 후 검색
hyde_prompt = PromptTemplate(
    input_variables=["question"],
    template="""질문: {question}

이 질문에 대한 상세한 답변을 작성하세요 (실제 사실 여부와 무관):"""
)

hyde_chain = LLMChain(llm=llm, prompt=hyde_prompt)

def hyde_search(question, retriever):
    # 1. LLM이 가상 답변 생성
    hypothetical_answer = hyde_chain.run(question=question)

    # 2. 가상 답변의 임베딩으로 검색
    docs = retriever.get_relevant_documents(hypothetical_answer)

    return docs

4.3 쿼리 재작성

# 쿼리 재작성 프롬프트
rewrite_prompt = PromptTemplate(
    input_variables=["question"],
    template="""다음 질문을 검색에 최적화된 형태로 재작성하세요.
핵심 키워드를 강조하고, 불필요한 단어는 제거하세요.

원래 질문: {question}

재작성된 질문:"""
)

rewrite_chain = LLMChain(llm=llm, prompt=rewrite_prompt)

def search_with_rewrite(question, retriever):
    rewritten = rewrite_chain.run(question=question)
    return retriever.get_relevant_documents(rewritten)

5. 프롬프트 엔지니어링

5.1 고급 시스템 프롬프트

from langchain.prompts import PromptTemplate

advanced_prompt = PromptTemplate(
    input_variables=["context", "question"],
    template="""당신은 정확하고 신뢰할 수 있는 AI 어시스턴트입니다.

지침:
1. 제공된 컨텍스트만을 사용하여 답변하세요
2. 컨텍스트에 답이 없으면 "제공된 정보에서 답을 찾을 수 없습니다"라고 답하세요
3. 추측하거나 외부 지식을 사용하지 마세요
4. 답변에 관련 섹션을 인용하세요
5. 간결하고 명확하게 답변하세요

컨텍스트:
{context}

질문: {question}

답변 (인용 포함):"""
)

5.2 Few-shot 프롬프트

few_shot_prompt = PromptTemplate(
    input_variables=["context", "question"],
    template="""다음 예제를 참고하여 답변하세요:

예제 1:
질문: RAG의 장점은?
답변: RAG의 주요 장점은 다음과 같습니다:
1. 최신 정보 활용 가능
2. 환각 현상 감소
[출처: RAG Overview, p.5]

예제 2:
질문: 벡터 데이터베이스란?
답변: 벡터 데이터베이스는 고차원 벡터를 효율적으로 저장하고 검색하는 데이터베이스입니다.
[출처: Vector DB Guide, p.12]

이제 다음 질문에 답하세요:

컨텍스트:
{context}

질문: {question}

답변:"""
)

5.3 Chain-of-Thought 프롬프트

cot_prompt = PromptTemplate(
    input_variables=["context", "question"],
    template="""단계별로 생각하며 답변하세요:

컨텍스트:
{context}

질문: {question}

답변 과정:
1. 먼저 질문의 핵심을 파악합니다:
2. 관련 정보를 컨텍스트에서 찾습니다:
3. 찾은 정보를 종합합니다:
4. 최종 답변:"""
)

6. 응답 생성 최적화

6.1 스트리밍 응답

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# 스트리밍으로 빠른 첫 응답
streaming_llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

qa_chain = RetrievalQA.from_chain_type(
    llm=streaming_llm,
    retriever=retriever
)

6.2 응답 캐싱

from langchain.cache import InMemoryCache, SQLiteCache
from langchain.globals import set_llm_cache

# 메모리 캐시
set_llm_cache(InMemoryCache())

# 또는 영구 캐시
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

# 동일한 질문에 대해 캐시된 답변 사용
result1 = qa_chain.run("RAG란?")  # LLM 호출
result2 = qa_chain.run("RAG란?")  # 캐시에서 반환 (빠름)

6.3 토큰 최적화

import tiktoken

def count_tokens(text, model="gpt-4"):
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

def optimize_context(docs, max_tokens=3000):
    """컨텍스트를 토큰 제한 내로 압축"""
    context = ""
    total_tokens = 0

    for doc in docs:
        doc_tokens = count_tokens(doc.page_content)
        if total_tokens + doc_tokens <= max_tokens:
            context += doc.page_content + "\n\n"
            total_tokens += doc_tokens
        else:
            break

    return context

7. 평가 및 모니터링

7.1 검색 품질 평가

from langchain.evaluation import load_evaluator

# 관련성 평가
relevance_evaluator = load_evaluator("criteria", criteria="relevance")

def evaluate_retrieval(query, retrieved_docs):
    scores = []
    for doc in retrieved_docs:
        result = relevance_evaluator.evaluate_strings(
            prediction=doc.page_content,
            input=query
        )
        scores.append(result['score'])

    return {
        'avg_score': sum(scores) / len(scores),
        'scores': scores
    }

# 평가 실행
eval_result = evaluate_retrieval(
    "RAG의 장점은?",
    retrieved_docs
)
print(f"평균 관련성 점수: {eval_result['avg_score']}")

7.2 답변 품질 평가

from langchain.evaluation.qa import QAEvalChain

# 테스트 데이터셋
test_questions = [
    {"question": "RAG란?", "answer": "Retrieval-Augmented Generation..."},
    {"question": "벡터 DB는?", "answer": "고차원 벡터를 저장..."},
]

# 평가 체인
eval_chain = QAEvalChain.from_llm(llm)

# 예측 및 평가
predictions = []
for item in test_questions:
    pred = qa_chain.run(item["question"])
    predictions.append({"query": item["question"], "result": pred})

graded_outputs = eval_chain.evaluate(
    test_questions,
    predictions,
    question_key="question",
    answer_key="answer",
    prediction_key="result"
)

for i, output in enumerate(graded_outputs):
    print(f"Q: {test_questions[i]['question']}")
    print(f"Grade: {output['results']}")

7.3 성능 모니터링

import time
from datetime import datetime

class RAGMonitor:
    def __init__(self):
        self.metrics = []

    def log_query(self, query, retrieval_time, generation_time, num_docs):
        self.metrics.append({
            'timestamp': datetime.now(),
            'query': query,
            'retrieval_time': retrieval_time,
            'generation_time': generation_time,
            'total_time': retrieval_time + generation_time,
            'num_docs': num_docs
        })

    def get_stats(self):
        if not self.metrics:
            return None

        return {
            'total_queries': len(self.metrics),
            'avg_retrieval_time': sum(m['retrieval_time'] for m in self.metrics) / len(self.metrics),
            'avg_generation_time': sum(m['generation_time'] for m in self.metrics) / len(self.metrics),
            'avg_total_time': sum(m['total_time'] for m in self.metrics) / len(self.metrics),
        }

# 사용 예
monitor = RAGMonitor()

def monitored_query(question):
    # 검색 시간 측정
    start = time.time()
    docs = retriever.get_relevant_documents(question)
    retrieval_time = time.time() - start

    # 생성 시간 측정
    start = time.time()
    answer = qa_chain.run(question)
    generation_time = time.time() - start

    # 로깅
    monitor.log_query(question, retrieval_time, generation_time, len(docs))

    return answer

# 통계 확인
stats = monitor.get_stats()
print(f"평균 검색 시간: {stats['avg_retrieval_time']:.3f}s")
print(f"평균 생성 시간: {stats['avg_generation_time']:.3f}s")

8. 고급 최적화 패턴

8.1 부모-자식 청킹

from langchain.storage import InMemoryStore
from langchain.retrievers import ParentDocumentRetriever

# 부모 문서 저장소
parent_store = InMemoryStore()

# 자식 청크로 검색, 부모 문서 반환
parent_retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=parent_store,
    child_splitter=RecursiveCharacterTextSplitter(chunk_size=400),
    parent_splitter=RecursiveCharacterTextSplitter(chunk_size=2000),
)

parent_retriever.add_documents(documents)

8.2 셀프 쿼리 (Self-Query)

from langchain.chains.query_constructor.base import AttributeInfo
from langchain.retrievers.self_query.base import SelfQueryRetriever

# 메타데이터 정의
metadata_field_info = [
    AttributeInfo(
        name="source",
        description="문서 출처",
        type="string",
    ),
    AttributeInfo(
        name="year",
        description="문서 작성 연도",
        type="integer",
    ),
]

# 자동으로 메타데이터 필터링
self_query_retriever = SelfQueryRetriever.from_llm(
    llm,
    vectorstore,
    "기술 문서 데이터베이스",
    metadata_field_info,
    verbose=True
)

# "2023년 이후 작성된 RAG 문서를 찾아줘" -> 자동 필터링
docs = self_query_retriever.get_relevant_documents(
    "2023년 이후 작성된 RAG 문서"
)

9. 비용 최적화

9.1 모델 선택 전략

# 간단한 질문 - 저렴한 모델
simple_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 복잡한 질문 - 고성능 모델
complex_llm = ChatOpenAI(model="gpt-4o", temperature=0)

def route_query(question):
    """질문 복잡도에 따라 모델 선택"""
    if len(question.split()) < 10:
        return simple_llm
    else:
        return complex_llm

9.2 임베딩 재사용

# 한 번만 임베딩, 여러 번 재사용
vectorstore.persist()  # 저장

# 나중에 로드 (재임베딩 불필요)
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

요약: 최적화 체크리스트

검색 품질

• 적절한 청크 크기 선택 (500-1500 토큰)
• 청크 오버랩 설정 (10-20%)
• 하이브리드 검색 적용
• MMR로 다양성 확보
• 재순위화 적용
• 메타데이터 필터링 활용

응답 품질

• 명확한 시스템 프롬프트
• Few-shot 예제 제공
• 출처 인용 요구
• 답변 평가 체계 구축

성능/비용

• 응답 캐싱
• 스트리밍 응답
• 적절한 모델 선택
• 임베딩 재사용
• 배치 처리