이 글은 누구를 위한 것인가
- RAG 시스템을 처음 구축하거나 개선하려는 팀
- RAG 품질을 수치로 측정하고 싶은 ML 엔지니어
- 멀티 홉 추론(여러 문서를 결합)이 필요한 개발자
들어가며
"우리 제품 FAQ를 AI가 답하게 하고 싶다." RAG는 이것의 기본이지만, 프로덕션에서 잘 작동하려면 청킹, 검색, 생성, 평가 전 단계를 신경 써야 한다.
이 글은 bluefoxdev.kr의 RAG 시스템 프로덕션 가이드 를 참고하여 작성했습니다.
1. RAG 시스템 아키텍처
[완전한 RAG 파이프라인]
오프라인 (인덱싱):
문서 수집 (PDF, 웹, DB)
→ 전처리 (텍스트 추출, 정제)
→ 청킹 (재귀/시맨틱)
→ 임베딩 (text-embedding-3-small)
→ 벡터 스토어 저장 (pgvector)
→ BM25 인덱스 업데이트
온라인 (쿼리):
사용자 질문
→ 쿼리 이해 (의도, 핵심어)
→ 하이브리드 검색 (벡터 + BM25)
→ 재랭킹 (cross-encoder)
→ 컨텍스트 구성
→ LLM 생성 (with citations)
→ 검증 (인용 일치 확인)
→ 응답 반환
[RAG 품질 지표 (RAGAS)]
Context Precision: 검색된 문서 관련도
Context Recall: 필요한 정보가 검색되는 비율
Faithfulness: 답변이 문서에 충실한 정도
Answer Relevance: 질문과 답변의 관련도
2. 프로덕션 RAG 구현
import anthropic
import json
from dataclasses import dataclass
client = anthropic.Anthropic()
@dataclass
class RAGResult:
answer: str
citations: list[dict]
retrieved_chunks: list[dict]
confidence: float
class ProductionRAG:
"""프로덕션 RAG 시스템"""
def __init__(self, vector_store, bm25_index):
self.vector_store = vector_store
self.bm25_index = bm25_index
async def retrieve(
self,
query: str,
top_k: int = 10,
filter_metadata: dict | None = None,
) -> list[dict]:
"""하이브리드 검색"""
import asyncio
# 병렬로 벡터/키워드 검색
vector_results, bm25_results = await asyncio.gather(
self.vector_store.search(query, k=top_k, filter=filter_metadata),
self.bm25_index.search(query, k=top_k),
)
# RRF 병합
scores = {}
k = 60
for rank, doc in enumerate(vector_results):
scores[doc["id"]] = scores.get(doc["id"], 0) + 1 / (k + rank + 1)
for rank, doc in enumerate(bm25_results):
scores[doc["id"]] = scores.get(doc["id"], 0) + 1 / (k + rank + 1)
all_docs = {d["id"]: d for d in vector_results + bm25_results}
sorted_docs = sorted(all_docs.values(), key=lambda d: scores.get(d["id"], 0), reverse=True)
return sorted_docs[:top_k]
async def generate_with_citations(
self,
question: str,
chunks: list[dict],
max_context_tokens: int = 6000,
) -> RAGResult:
"""인용 포함 답변 생성"""
# 컨텍스트 구성
context = ""
used_chunks = []
token_count = 0
for chunk in chunks:
chunk_tokens = len(chunk["text"]) // 3 # 대략적 추정
if token_count + chunk_tokens > max_context_tokens:
break
context += f"\n[출처 {len(used_chunks)+1}] {chunk['text']}\n"
used_chunks.append(chunk)
token_count += chunk_tokens
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1500,
messages=[{
"role": "user",
"content": f"""다음 출처 문서들을 바탕으로 질문에 답하세요.
출처에 없는 내용은 말하지 마세요.
답변에 [출처 N] 형식으로 인용을 표시하세요.
{context}
질문: {question}
JSON으로 답하세요:
{{
"answer": "인용이 포함된 답변 [출처 1] ...",
"citations": [
{{"source_number": 1, "quoted_text": "인용 원문 일부", "relevance": "어떻게 사용됐는지"}}
],
"confidence": 0.0-1.0,
"sufficient_context": true/false
}}"""
}]
)
result = json.loads(response.content[0].text)
return RAGResult(
answer=result["answer"],
citations=result["citations"],
retrieved_chunks=used_chunks,
confidence=result["confidence"],
)
async def multi_hop_rag(
self,
complex_question: str,
max_hops: int = 3,
) -> RAGResult:
"""멀티 홉 추론: 여러 단계 검색"""
all_chunks = []
current_question = complex_question
for hop in range(max_hops):
chunks = await self.retrieve(current_question, top_k=5)
new_chunks = [c for c in chunks if c["id"] not in {ch["id"] for ch in all_chunks}]
all_chunks.extend(new_chunks)
if not new_chunks:
break
# 다음 검색을 위한 서브 질문 생성
context_preview = "\n".join(c["text"][:200] for c in new_chunks[:3])
sub_question_response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=200,
messages=[{
"role": "user",
"content": f"""원래 질문: {complex_question}
지금까지 찾은 정보: {context_preview}
아직 답을 위해 더 찾아야 할 것이 있다면 서브 질문 하나를 제시하세요.
없으면 "충분"이라고만 답하세요."""
}]
)
next_q = sub_question_response.content[0].text.strip()
if "충분" in next_q:
break
current_question = next_q
return await self.generate_with_citations(complex_question, all_chunks)
async def evaluate_rag_quality(
questions: list[str],
ground_truths: list[str],
rag_system: ProductionRAG,
) -> dict:
"""RAG 품질 평가 (RAGAS 지표)"""
faithfulness_scores = []
relevance_scores = []
for question, truth in zip(questions, ground_truths):
chunks = await rag_system.retrieve(question)
result = await rag_system.generate_with_citations(question, chunks)
# Faithfulness: 답변이 출처에 충실한가
eval_response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=100,
messages=[{
"role": "user",
"content": f"""답변이 출처 문서에만 근거하는지 0-1 점수로 평가:
출처: {' '.join(c['text'][:200] for c in result.retrieved_chunks[:3])}
답변: {result.answer}
JSON: {{"faithfulness": 0.0-1.0}}"""
}]
)
f_score = json.loads(eval_response.content[0].text)["faithfulness"]
faithfulness_scores.append(f_score)
return {
"avg_faithfulness": sum(faithfulness_scores) / len(faithfulness_scores),
"n_evaluated": len(questions),
}
마무리
RAG 품질의 병목은 대부분 검색 단계에 있다. 답변이 이상하면 먼저 검색된 청크를 확인하라. RAGAS 지표로 Faithfulness가 낮으면 프롬프트를 강화하고, Context Recall이 낮으면 청킹이나 검색을 개선하라. 인용을 강제하면 환각이 줄고 사용자 신뢰가 높아진다.