RAG 심화 기법¶
Retrieval-Augmented Generation의 고급 기법과 최적화 전략.
개요¶
기본 RAG를 넘어 실제 프로덕션에서 성능을 높이기 위한 심화 기법들을 다룬다. 검색 품질, 컨텍스트 최적화, 답변 정확도 향상에 초점을 맞춘다.
RAG 진화 단계¶
검색 최적화¶
Query Transformation¶
사용자 쿼리를 검색에 최적화된 형태로 변환.
| 기법 | 설명 | 효과 |
|---|---|---|
| Query Rewriting | LLM으로 쿼리 재작성 | 모호함 해소 |
| Query Expansion | 동의어/관련어 추가 | 재현율 향상 |
| HyDE | 가상 문서 생성 후 검색 | 의미적 매칭 |
| Step-back Prompting | 더 넓은 개념으로 쿼리 | 컨텍스트 확장 |
# HyDE (Hypothetical Document Embedding)
def hyde_retrieval(query, llm, retriever):
# 1. 가상 답변 생성
hypothetical_doc = llm.generate(
f"Write a passage that answers: {query}"
)
# 2. 가상 답변으로 검색
results = retriever.search(hypothetical_doc)
return results
Multi-Query Retrieval¶
단일 쿼리를 여러 관점으로 분해.
def multi_query_retrieval(query, llm, retriever, k=3):
# 다양한 쿼리 생성
prompt = f"""Generate {k} different versions of this question
to retrieve relevant documents: {query}"""
queries = llm.generate(prompt).split('\n')
# 각 쿼리로 검색 후 병합
all_docs = []
for q in queries:
docs = retriever.search(q)
all_docs.extend(docs)
# 중복 제거 및 re-ranking
return deduplicate_and_rank(all_docs)
Hybrid Search¶
키워드 검색과 의미 검색의 결합.
def hybrid_search(query, alpha=0.5):
# BM25 (키워드)
bm25_scores = bm25_search(query)
# Dense (의미)
dense_scores = dense_search(query)
# 가중 결합
final_scores = alpha * bm25_scores + (1 - alpha) * dense_scores
return final_scores
가중치 가이드: | 상황 | alpha (BM25 비중) | |------|-------------------| | 전문 용어 중심 | 0.7 | | 일반 질문 | 0.3 | | 균형 | 0.5 |
컨텍스트 최적화¶
Chunking 전략¶
| 전략 | 설명 | 적합한 상황 |
|---|---|---|
| Fixed-size | 고정 토큰 수 | 균일한 문서 |
| Sentence | 문장 단위 | 짧은 답변 |
| Paragraph | 문단 단위 | 완결된 맥락 |
| Semantic | 의미 단위 | 주제 전환 많음 |
| Recursive | 계층적 분할 | 긴 문서 |
# Semantic Chunking
from langchain.text_splitter import SemanticChunker
splitter = SemanticChunker(
embeddings=embeddings,
breakpoint_threshold_type="percentile",
breakpoint_threshold_amount=95
)
chunks = splitter.split_text(document)
Parent Document Retrieval¶
작은 청크로 검색, 큰 청크로 컨텍스트 제공.
문서
├── Parent Chunk (2000 tokens) ← 컨텍스트로 전달
│ ├── Child Chunk (200 tokens) ← 검색 대상
│ └── Child Chunk (200 tokens)
└── Parent Chunk
└── ...
Contextual Compression¶
검색된 문서에서 관련 부분만 추출.
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=retriever
)
Re-ranking¶
검색 후 순위 재조정으로 정밀도 향상.
Cross-Encoder Re-ranking¶
from sentence_transformers import CrossEncoder
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
def rerank(query, documents, top_k=5):
pairs = [[query, doc.page_content] for doc in documents]
scores = reranker.predict(pairs)
# 점수로 정렬
ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
return [doc for doc, score in ranked[:top_k]]
Cohere Rerank¶
import cohere
co = cohere.Client(api_key)
results = co.rerank(
query=query,
documents=[doc.page_content for doc in docs],
top_n=5,
model='rerank-english-v2.0'
)
LLM-based Re-ranking¶
def llm_rerank(query, documents, llm):
prompt = f"""Given the question: {query}
Rank these documents by relevance (most relevant first):
{enumerate_documents(documents)}
Return only the document numbers in order of relevance."""
ranking = llm.generate(prompt)
return reorder_by_ranking(documents, ranking)
Self-RAG¶
검색 필요성과 답변 품질을 스스로 판단.
def self_rag(query, llm, retriever):
# 1. 검색 필요성 판단
needs_retrieval = llm.generate(
f"Does answering '{query}' require external knowledge? (yes/no)"
)
if needs_retrieval == "no":
return llm.generate(query)
# 2. 검색
docs = retriever.search(query)
# 3. 관련성 필터링
relevant_docs = []
for doc in docs:
is_relevant = llm.generate(
f"Is this relevant to '{query}'? {doc.content} (yes/no)"
)
if is_relevant == "yes":
relevant_docs.append(doc)
# 4. 답변 생성
answer = llm.generate(query, context=relevant_docs)
# 5. 답변 검증
is_supported = llm.generate(
f"Is this answer supported by the documents? {answer}"
)
return answer if is_supported else regenerate(query, relevant_docs)
답변 품질 향상¶
Citation/Attribution¶
답변에 출처 표시.
def generate_with_citations(query, docs, llm):
prompt = f"""Based on the following sources, answer the question.
Include [N] citations after each claim.
Sources:
{format_sources_with_numbers(docs)}
Question: {query}
"""
return llm.generate(prompt)
Hallucination Detection¶
def check_hallucination(answer, sources, llm):
claims = extract_claims(answer)
for claim in claims:
is_supported = llm.generate(
f"Is this claim supported by the sources? "
f"Claim: {claim}\nSources: {sources}"
)
if is_supported == "no":
return True, claim
return False, None
평가 지표¶
Retrieval 평가¶
| 지표 | 설명 | 계산 |
|---|---|---|
| Recall@k | 상위 k개에 정답 포함 | 정답포함/전체 |
| MRR | 첫 정답 순위의 역수 | 1/rank |
| NDCG@k | 순위 가중 정확도 | - |
| Hit Rate | 정답 검색 여부 | - |
Generation 평가¶
| 지표 | 측정 대상 |
|---|---|
| Faithfulness | 출처에 기반한 답변인가 |
| Relevance | 질문에 적절한 답변인가 |
| Correctness | 사실적으로 정확한가 |
# RAGAS 평가
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision
result = evaluate(
dataset,
metrics=[faithfulness, answer_relevancy, context_precision]
)
실무 체크리스트¶
검색 품질 개선¶
- [ ] Embedding 모델 선택 (도메인 특화 vs 범용)
- [ ] Chunking 크기/전략 실험
- [ ] Hybrid search 비율 조정
- [ ] Re-ranking 적용
컨텍스트 최적화¶
- [ ] 컨텍스트 길이 vs 품질 트레이드오프
- [ ] Parent document retrieval 고려
- [ ] Contextual compression 적용
답변 품질¶
- [ ] Citation 요구
- [ ] Hallucination 체크
- [ ] Self-RAG 패턴 적용
참고 자료¶
논문¶
- "Retrieval-Augmented Generation for Large Language Models: A Survey" (2024)
- "Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflection" (2023)
- "CRAG: Corrective Retrieval Augmented Generation" (2024)