검색, 평가, 그리고 신뢰할 수 있는 시스템
소규모 기업을 위한 제대로 된 RAG 시스템 [3/4]
2편에서 데이터 수집 파이프라인을 완성했습니다. 문서를 전처리하고, 청킹하고, Anthropic의 Contextual Retrieval로 문맥을 붙이고, Ollama 임베딩으로 벡터화해서 ChromaDB에 저장하는 과정이었습니다. python pipeline.py --input-dir ./documents 한 줄이면 사내 문서가 검색 가능한 상태로 인덱싱됩니다.
그런데 데이터를 준비한 것만으로 RAG 시스템이 완성되지는 않습니다. 사용자 질문을 받아서 관련 문서를 찾고, 그 문서를 기반으로 답변을 만들어내는 검색과 생성 계층이 빠져 있거든요. 그 답변이 정말 맞는 건지 검증하는 체계도 필요하고, 시스템이 프로덕션에서 조용히 망가지고 있지는 않은지 감시하는 관측 체계도 있어야 합니다.
현업 개발자에게 RAG에서 가장 중요한 단계가 뭐냐고 물어보면, 의외로 많은 사람이 "검색"이라고 답합니다. Towards Data Science의 프로덕션 RAG 교훈 시리즈에서도 같은 결론을 냈는데, "RAG 프로젝트의 60%는 RAG가 필요한 것이 아니라 더 나은 검색이 필요합니다"라는 말이 핵심이었습니다. LLM은 주어진 컨텍스트가 좋으면 좋은 답변을 만들고, 컨텍스트가 나쁘면 나쁜 답변을 만듭니다. 검색 품질이 곧 답변 품질입니다.
이번 편에서는 세 가지를 다룹니다. 첫째, 검색 품질을 높이는 실전 기법입니다. 하이브리드 검색과 리랭킹으로 벡터 검색만 쓸 때보다 정확도를 30% 이상 끌어올립니다. 둘째, 답변의 신뢰성을 보장하는 가드레일 시스템입니다. DoorDash가 수십만 건의 일일 지원 요청에 적용해서 할루시네이션을 90% 줄인 패턴을 소규모 기업 환경에 맞게 구현합니다. 셋째, 시스템이 제대로 작동하는지 측정하는 평가 체계입니다. RAGAS 프레임워크로 검색 품질과 답변 충실도를 정량적으로 측정하고, Langfuse로 프로덕션 운영을 모니터링합니다.
2편에서 만든 데이터 수집 파이프라인 위에 이번 편의 검색 파이프라인을 얹으면, 소규모 기업이 실제로 돌릴 수 있는 프로덕션급 RAG 시스템이 됩니다. 모든 코드는 Ollama 로컬 실행 기준이므로 API 비용은 사실상 무료입니다.
다음 그림은 이번 편에서 구축할 전체 검색 파이프라인의 흐름입니다. 사용자 질문이 들어오면 하이브리드 검색, 리랭킹, LLM 생성, 가드레일을 거쳐 신뢰할 수 있는 답변으로 나옵니다.
질문부터 답변까지 6단계를 거칩니다. 하이브리드 검색으로 후보를 넓게 잡고, 크로스인코더 리랭킹으로 정밀하게 추린 뒤, Ollama LLM이 답변을 생성합니다. 생성된 답변은 2단계 가드레일을 통과해야 사용자에게 전달되고, RAGAS 평가와 Langfuse 모니터링이 전체 시스템의 품질을 지속적으로 측정합니다.
하이브리드 검색 - 벡터와 키워드의 결합
1편에서 벡터 검색만으로는 부족하다는 점을 언급했습니다. 벡터 검색은 "의미적 유사성"만 보기 때문에, 사용자가 "에러 코드 E1045"를 검색하면 "에러"라는 의미와 유사한 모든 문서를 가져옵니다. 정작 "E1045"라는 정확한 코드는 벡터 유사도에서 잘 반영되지 않습니다.
반대로, 전통적인 키워드 검색(BM25)은 정확한 용어 매칭에 강합니다. "E1045"를 검색하면 정확히 그 문자열이 들어있는 문서를 찾아냅니다. 하지만 의미적으로 관련 있지만 다른 단어를 쓴 문서는 놓칩니다. "에러 E1045는 메모리 부족으로 발생합니다"를 찾을 수 있어도, 같은 문제를 "OOM 이슈"로 설명한 문서는 못 찾습니다.
하이브리드 검색은 두 방식을 결합합니다. 벡터 검색으로 의미적으로 관련 있는 문서를 넓게 잡고, BM25로 키워드가 정확히 일치하는 문서를 잡은 뒤, 두 결과를 합치는 겁니다. Superlinked VectorHub의 분석에 따르면 하이브리드 검색은 단일 방식 대비 정확도가 15~30% 올라갑니다. Enterprise RAG Architecture 가이드에서도 하이브리드 검색을 "프로덕션 표준"이라고 부르고요.
두 검색 결과를 합치는 방법이 문제입니다. 벡터 검색은 코사인 유사도 점수(0~1)를 반환하고, BM25는 전혀 다른 스코어링 체계를 씁니다. 점수 스케일이 다르기 때문에 단순히 점수를 더하거나 평균 낼 수 없습니다.
Reciprocal Rank Fusion이 이 문제를 해결합니다. 각 검색 결과의 순위(rank)를 기준으로 점수를 계산합니다. 공식은 간단합니다. 문서 d의 RRF 점수는 각 검색 결과 목록에서 해당 문서의 순위에 상수 k를 더한 뒤, 그 역수를 모두 합한 값입니다. k는 보통 60을 씁니다. 벡터 검색에서 1등이고 BM25에서 5등인 문서의 점수는 1/(60+1) + 1/(60+5) = 0.0164 + 0.0154 = 0.0318입니다. 벡터 검색에서 3등이고 BM25에서 1등인 문서는 1/(60+3) + 1/(60+1) = 0.0159 + 0.0164 = 0.0322로 약간 더 높습니다. 두 검색 방식에서 모두 상위권에 오른 문서가 자연스럽게 최상위로 올라갑니다.
코드로 구현하겠습니다. 2편에서 만든 ChromaDB 벡터 저장소 위에 BM25 검색을 추가하고, RRF로 결과를 합칩니다.
# bm25_retriever.py """BM25 키워드 검색기.""" import math import re from collections import Counter from dataclasses import dataclass @dataclass class BM25Document: doc_id: str content: str metadata: dict class BM25Retriever: """BM25 알고리즘 기반 키워드 검색기. rank-bm25 라이브러리를 써도 되지만, 원리를 이해하기 위해 직접 구현합니다. 프로덕션에서는 Elasticsearch나 Meilisearch의 BM25를 쓰는 것이 성능상 유리합니다. """ def __init__(self, k1: float = 1.5, b: float = 0.75): self.k1 = k1 # 용어 빈도 포화 파라미터 self.b = b # 문서 길이 정규화 파라미터 self.documents: list[BM25Document] = [] self.doc_lengths: list[int] = [] self.avg_doc_length: float = 0.0 self.doc_freqs: dict[str, int] = {} # 각 단어가 등장하는 문서 수 self.doc_term_freqs: list[dict[str, int]] = [] # 문서별 단어 빈도 self.n_docs: int = 0 def _tokenize(self, text: str) -> list[str]: """간단한 토크나이저. 한국어와 영어를 모두 처리합니다.""" # 소문자 변환 후 알파벳/숫자/한글 단위로 분리 text = text.lower() tokens = re.findall(r'[a-z0-9]+|[\uac00-\ud7af]+', text) return tokens def index(self, documents: list[BM25Document]): """문서 컬렉션을 인덱싱합니다.""" self.documents = documents self.n_docs = len(documents) self.doc_term_freqs = [] self.doc_lengths = [] for doc in documents: tokens = self._tokenize(doc.content) self.doc_lengths.append(len(tokens)) term_freq = Counter(tokens) self.doc_term_freqs.append(dict(term_freq)) for term in set(tokens): self.doc_freqs[term] = self.doc_freqs.get(term, 0) + 1 self.avg_doc_length = sum(self.doc_lengths) / max(self.n_docs, 1) def search(self, query: str, top_k: int = 20) -> list[tuple[str, float]]: """BM25 점수 기준으로 상위 k개 문서를 반환합니다. Returns: [(doc_id, score), ...] """ query_tokens = self._tokenize(query) scores = [] for i, doc in enumerate(self.documents): score = 0.0 doc_len = self.doc_lengths[i] term_freqs = self.doc_term_freqs[i] for term in query_tokens: if term not in term_freqs: continue tf = term_freqs[term] df = self.doc_freqs.get(term, 0) idf = math.log((self.n_docs - df + 0.5) / (df + 0.5) + 1) numerator = tf * (self.k1 + 1) denominator = tf + self.k1 * (1 - self.b + self.b * doc_len / self.avg_doc_length) score += idf * numerator / denominator if score > 0: scores.append((doc.doc_id, score)) scores.sort(key=lambda x: x[1], reverse=True) return scores[:top_k]
BM25는 직접 구현했지만, 한 가지 짚고 넘어갈 한계가 있습니다. _tokenize 메서드가 한국어를 공백과 정규식으로 분리하는데, 이 방식으로는 조사가 붙은 단어를 처리하지 못합니다. "인덱스"를 검색하면 "인덱스를", "인덱스의" 같은 형태와 매칭되지 않습니다. 한국어 BM25가 제대로 작동하려면 형태소 분석이 필수입니다.
프로덕션에서는 두 가지 선택지가 있습니다. 첫째, kiwipiepy를 토크나이저로 교체하는 방법입니다. kiwipiepy는 C++ 기반의 한국어 형태소 분석기로, pip 설치 한 줄이면 되고 별도 시스템 의존성이 없습니다.
# 한국어 형태소 분석기를 적용한 토크나이저 from kiwipiepy import Kiwi kiwi = Kiwi() def tokenize_korean(text: str) -> list[str]: """kiwipiepy로 한국어 형태소 분석 후 명사/동사 어근만 추출합니다.""" tokens = [] for token in kiwi.tokenize(text): # NNG(일반명사), NNP(고유명사), VV(동사), VA(형용사), SL(영어) if token.tag in ("NNG", "NNP", "VV", "VA", "SL"): tokens.append(token.form.lower()) return tokens
이렇게 하면 "데이터베이스의 인덱스를 최적화했습니다"에서 "데이터베이스", "인덱스", "최적화"를 정확히 뽑아냅니다. 설치는 pip install kiwipiepy입니다.
둘째, Elasticsearch의 nori 분석기를 쓰는 방법입니다. 이미 Elasticsearch를 운영하고 있다면 nori 플러그인을 설치해서 BM25와 한국어 형태소 분석을 한번에 해결할 수 있습니다. 앞서 구현한 BM25 코드는 알고리즘의 원리를 이해하기 위한 것이고, 프로덕션에서는 Elasticsearch나 Meilisearch 같은 전문 검색 엔진의 BM25를 쓰는 것이 성능과 확장성 면에서 유리합니다.
다음은 벡터 검색과 BM25 결과를 Reciprocal Rank Fusion으로 합치는 코드입니다.
# hybrid_retriever.py """하이브리드 검색: 벡터 + BM25 + Reciprocal Rank Fusion.""" from dataclasses import dataclass from bm25_retriever import BM25Retriever, BM25Document @dataclass class SearchResult: doc_id: str content: str metadata: dict score: float sources: list[str] # ["vector", "bm25"] 등 class HybridRetriever: """벡터 검색과 BM25 검색을 RRF로 결합합니다.""" def __init__( self, vector_store, embedding_service, bm25_retriever: BM25Retriever, rrf_k: int = 60, vector_weight: float = 1.0, bm25_weight: float = 1.0, ): self.vector_store = vector_store self.embedding = embedding_service self.bm25 = bm25_retriever self.rrf_k = rrf_k self.vector_weight = vector_weight self.bm25_weight = bm25_weight def search(self, query: str, top_k: int = 10, initial_k: int = 30) -> list[SearchResult]: """하이브리드 검색을 수행합니다. Args: query: 사용자 질문 top_k: 최종 반환할 문서 수 initial_k: 각 검색 방식에서 가져올 초기 문서 수 """ # 1. 벡터 검색 query_embedding = self.embedding.embed(query) vector_results = self.vector_store.search(query_embedding, n_results=initial_k) # 2. BM25 검색 bm25_results = self.bm25.search(query, top_k=initial_k) # 3. RRF 병합 rrf_scores: dict[str, float] = {} doc_contents: dict[str, str] = {} doc_metadata: dict[str, dict] = {} doc_sources: dict[str, list[str]] = {} # 벡터 검색 결과에 RRF 점수 부여 for rank, result in enumerate(vector_results): doc_id = result["id"] rrf_score = self.vector_weight / (self.rrf_k + rank + 1) rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + rrf_score doc_contents[doc_id] = result["content"] doc_metadata[doc_id] = result.get("metadata", {}) doc_sources.setdefault(doc_id, []).append("vector") # BM25 검색 결과에 RRF 점수 부여 for rank, (doc_id, _) in enumerate(bm25_results): rrf_score = self.bm25_weight / (self.rrf_k + rank + 1) rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + rrf_score doc_sources.setdefault(doc_id, []).append("bm25") # BM25에서만 검색된 문서의 content 보강 if doc_id not in doc_contents: bm25_doc = next( (d for d in self.bm25.documents if d.doc_id == doc_id), None ) if bm25_doc: doc_contents[doc_id] = bm25_doc.content doc_metadata[doc_id] = bm25_doc.metadata # 4. RRF 점수 기준 정렬 sorted_docs = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True) results = [] for doc_id, score in sorted_docs[:top_k]: results.append(SearchResult( doc_id=doc_id, content=doc_contents.get(doc_id, ""), metadata=doc_metadata.get(doc_id, {}), score=score, sources=doc_sources.get(doc_id, []), )) return results
HybridRetriever의 search 메서드를 보면, 벡터 검색과 BM25를 각각 실행한 뒤 RRF로 합치는 흐름이 명확합니다. initial_k=30은 각 검색 방식에서 30개를 넓게 가져온다는 뜻입니다. 최종적으로 top_k=10개만 반환합니다. 넓게 잡은 뒤 좁히는 방식이 RRF의 효과를 극대화합니다.
vector_weight와 bm25_weight로 두 검색 방식의 비중을 조절할 수 있습니다. 기술 문서처럼 정확한 용어가 중요한 도메인에서는 bm25_weight를 높이고, 자연어 질문이 많은 환경에서는 vector_weight를 높이는 식으로 튜닝합니다. 기본값 1.0:1.0에서 시작해서, 실제 쿼리 로그를 보면서 조정하는 것이 현실적인 접근법입니다.
하이브리드 검색이 실제로 얼마나 효과가 있는지 살펴보겠습니다. Superlinked VectorHub의 벤치마크에서는 하이브리드 검색(BM25 + 벡터)이 벡터 검색 단독 대비 recall을 15~30% 끌어올렸습니다. Enterprise RAG Architecture 가이드에서도 하이브리드 검색 도입을 프로덕션 RAG의 Phase 1으로 잡고, 1~2주면 구현 가능하다고 권장하고 있습니다.
특히 효과가 두드러지는 경우가 있습니다. 에러 코드, API 엔드포인트, 설정값 같은 정확한 문자열을 검색할 때 BM25가 빛을 발합니다. 벡터 검색은 "서버가 느릴 때 어떻게 하나요?"처럼 의미적 질문에서 강합니다. 사내 문서 검색에서는 두 유형의 질문이 섞여서 들어오기 때문에, 하이브리드가 단일 방식보다 거의 항상 낫습니다.
리랭킹 - 후보를 추리는 두 번째 필터
하이브리드 검색으로 가져온 문서가 10개라고 합시다. 이 10개가 모두 같은 수준으로 관련 있는 것은 아닙니다. 어떤 문서는 질문에 정확히 답하는 내용이고, 어떤 문서는 비슷한 주제지만 질문과는 미묘하게 다른 내용입니다. 이 차이를 구분해서 가장 관련 있는 문서를 위로 올리는 작업이 리랭킹입니다.
리랭킹을 이해하려면 bi-encoder와 cross-encoder의 차이를 알아야 합니다. 벡터 검색에 쓰는 임베딩 모델은 bi-encoder입니다. 쿼리와 문서를 각각 독립적으로 벡터로 변환한 뒤, 두 벡터 사이의 거리를 비교합니다. 빠르지만, 쿼리와 문서 사이의 미묘한 상호작용을 놓칩니다.
cross-encoder는 다릅니다. 쿼리와 문서를 하나의 입력으로 합쳐서 BERT 같은 트랜스포머 모델에 통째로 넣습니다. 모델이 쿼리의 모든 토큰과 문서의 모든 토큰 사이의 관계를 직접 계산합니다. 정확도가 높지만, 문서 하나당 한 번씩 모델을 돌려야 하기 때문에 느립니다. 10개 문서를 리랭킹하는 데 200~500ms가 추가됩니다.
이 때문에 리랭킹은 "1단계 검색으로 후보를 넓게 잡고, 2단계 리랭킹으로 후보를 정밀하게 추리는" 2단계 구조로 씁니다. 1단계에서 30~50개를 빠르게 가져오고, 2단계에서 상위 5~10개로 추립니다.
다음 그림은 이 2단계 검색 구조를 보여줍니다. 1단계에서 하이브리드 검색이 넓게 후보를 잡고, 2단계에서 크로스인코더가 쿼리와 문서를 함께 분석하여 가장 관련성 높은 문서만 선별합니다.
핵심은 "Recall 먼저, 그 다음 Precision"입니다. bi-encoder는 쿼리와 문서를 각각 독립적으로 인코딩하여 빠르지만 미묘한 관계를 놓칠 수 있고, cross-encoder는 쿼리와 문서를 함께 넣어 정확하지만 느립니다. 두 방식의 장점을 순차적으로 활용하는 것이 2단계 검색의 핵심입니다.
리랭킹 모델은 두 가지 선택지가 있습니다. 오픈소스냐, API냐. Zero Entropy의 2026년 리랭킹 모델 비교 가이드에 따르면, Cohere Rerank과 ms-marco-MiniLM-L-6-v2가 정확도와 속도의 균형이 가장 좋습니다. 소규모 기업에서는 ms-marco-MiniLM을 로컬에서 돌리는 것이 비용 면에서 유리하고, API 비용을 감수할 수 있다면 Cohere Rerank이 정확도가 더 높습니다.
하나 중요한 점이 있습니다. bswen의 하이브리드 검색 vs 리랭커 분석에서 강조하는 건데, 리랭킹은 recall이 충분히 확보된 후에 추가해야 합니다. recall@50이 90% 미만이면 리랭킹을 추가해도 효과가 미미합니다. 1단계 검색에서 관련 문서를 제대로 못 찾고 있는데, 못 찾은 결과의 순서를 재배열해봐야 소용이 없기 때문입니다. 먼저 하이브리드 검색으로 recall을 90% 이상 확보한 뒤, precision을 위해 리랭킹을 추가하는 순서가 맞습니다.
코드를 보겠습니다. sentence-transformers 라이브러리의 CrossEncoder를 사용합니다.
# reranker.py """크로스인코더 기반 리랭커.""" from dataclasses import dataclass from sentence_transformers import CrossEncoder @dataclass class RankedResult: doc_id: str content: str metadata: dict original_score: float rerank_score: float sources: list[str] class Reranker: """크로스인코더로 검색 결과를 재순위화합니다.""" def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"): self.model = CrossEncoder(model_name) def rerank( self, query: str, results: list, top_k: int = 5, ) -> list[RankedResult]: """검색 결과를 크로스인코더로 재순위화합니다. Args: query: 사용자 질문 results: HybridRetriever.search()의 반환값 top_k: 최종 반환할 문서 수 """ if not results: return [] # 쿼리-문서 쌍 구성 pairs = [(query, r.content) for r in results] # 크로스인코더 점수 계산 scores = self.model.predict(pairs) # 결과 조합 ranked = [] for result, score in zip(results, scores): ranked.append(RankedResult( doc_id=result.doc_id, content=result.content, metadata=result.metadata, original_score=result.score, rerank_score=float(score), sources=result.sources, )) # 리랭크 점수로 재정렬 ranked.sort(key=lambda x: x.rerank_score, reverse=True) return ranked[:top_k]
ms-marco-MiniLM-L-6-v2는 모델 크기가 작아서(~80MB) CPU에서도 10개 문서를 200ms 안에 리랭킹합니다. GPU가 있으면 50ms 이내로 줄어듭니다. 소규모 기업 서버에서 부담 없이 돌릴 수 있는 크기입니다.
Enterprise RAG Architecture 가이드의 벤치마크에 따르면, 하이브리드 검색에 크로스인코더 리랭킹을 얹으면 정확도가 23% 더 올라갑니다. 하이브리드 검색 자체가 15~30%이니, 벡터 검색만 쓰던 시스템 기준으로 전체적으로 40~50% 정도 정확도가 좋아지는 셈입니다.
설치에 필요한 패키지는 다음과 같습니다.
pip install sentence-transformers torch
모델은 첫 실행 시 자동으로 다운로드됩니다. 오프라인 환경에서는 미리 받아서 로컬 경로를 지정할 수 있습니다.
쿼리 변환 - HyDE로 검색 품질 한 단계 올리기
하이브리드 검색과 리랭킹만으로도 검색 품질이 크게 올라가지만, 한 가지 더 고려할 기법이 있습니다. 사용자의 질문 자체를 변환해서 검색 효과를 높이는 HyDE입니다.
HyDE의 원리는 이렇습니다. 사용자가 "서버 메모리 부족하면 어떻게 해?"라고 짧게 물었다고 합시다. 이 질문의 임베딩은 짧은 질문형 문장이라 문서의 임베딩과 공간적으로 거리가 있을 수 있습니다. HyDE는 LLM에게 "이 질문에 대한 가상의 답변 문서를 만들어달라"고 요청합니다. LLM이 만든 가상 답변은 "서버 메모리 부족 시 OOM Killer가 프로세스를 종료합니다. 해결 방법은 swap 영역 확대, 메모리 누수 점검, JVM heap 크기 조정 등이 있습니다..."처럼 실제 문서와 비슷한 형태가 됩니다. 이 가상 답변의 임베딩으로 벡터 검색을 하면, 짧은 질문으로 검색하는 것보다 관련 문서를 더 잘 찾아냅니다.
# hyde.py """HyDE(Hypothetical Document Embeddings) 쿼리 변환.""" class HyDE: """사용자 질문을 가상의 답변 문서로 변환하여 검색 품질을 높입니다.""" def __init__(self, llm_client, embedding_service): self.llm = llm_client self.embedding = embedding_service def transform_query(self, query: str) -> list[float]: """질문을 가상 답변으로 변환한 뒤 임베딩을 반환합니다.""" hypothetical_doc = self.llm.generate( f"다음 질문에 대해 사내 문서에서 찾을 수 있을 법한 답변을 " f"200자 내외로 작성하세요. 정확하지 않아도 됩니다.\n\n" f"질문: {query}\n\n답변:" ) return self.embedding.embed(hypothetical_doc)
코드가 짧습니다. LLM 호출 한 번, 임베딩 한 번이 전부입니다. 다만 LLM 호출이 추가되므로 응답 시간이 1~3초 늘어납니다. 모든 쿼리에 적용하기보다는, 질문이 짧거나(5단어 이하) 초기 검색 결과의 confidence가 낮을 때 선택적으로 적용하는 것이 현실적입니다.
1편의 아키텍처에서도 쿼리 변환을 포함했었는데, 이번 시스템에서는 선택적 적용을 권장합니다. 하이브리드 검색 + 리랭킹이 이미 충분한 정확도를 확보한 상태에서, HyDE는 "짧은 질문"이나 "추상적 질문"에 대한 추가 보험 역할입니다.
답변 생성과 가드레일
검색으로 관련 문서를 찾았으면, 이제 LLM에게 그 문서를 기반으로 답변을 생성하게 합니다. 이 과정이 단순해 보이지만, 프로덕션에서는 "답변이 검색된 컨텍스트에 충실한가"를 보장하는 것이 가장 어려운 문제입니다.
LLM은 주어진 컨텍스트에 없는 내용도 자신있게 만들어냅니다. 검색된 문서에 "2024년 매출은 10억입니다"라고 적혀 있는데, LLM이 "2024년 매출은 10억이며, 전년 대비 15% 증가한 수치입니다"라고 답할 수 있습니다. "전년 대비 15% 증가"라는 정보는 컨텍스트에 없었는데 LLM이 그럴듯하게 덧붙인 겁니다. 이것이 RAG 환경에서의 할루시네이션입니다.
DoorDash의 Dasher 지원 시스템 사례가 이 문제를 해결하는 좋은 참고 모델입니다. DoorDash는 수십만 건의 일일 지원 요청을 처리하는 RAG 기반 챗봇을 운영하면서, 2단계 가드레일 시스템을 적용해 할루시네이션을 90% 줄이고 심각한 규정 위반을 99% 감소시켰습니다.
DoorDash의 2단계 가드레일 패턴은 이렇습니다. 1단계는 저비용 시맨틱 유사도 검사입니다. 생성된 답변과 검색된 컨텍스트 사이의 시맨틱 유사도를 계산해서, 임계값 이상이면 통과, 미만이면 2단계로 넘기는 거죠. 빠르고 비용이 거의 없습니다. 2단계는 LLM 기반 심층 검증입니다. 별도의 LLM에게 "이 답변이 주어진 컨텍스트에 근거하고 있는가?", "정책을 위반하는 내용이 있는가?"를 판단하게 합니다. 느리고 비용이 들지만 정확하고요.
이 구조의 핵심은 비용 효율성입니다. 대부분의 답변은 1단계에서 빠르게 통과하고, 의심스러운 소수만 2단계에서 정밀 검증하는 방식이거든요. 모든 답변에 LLM judge를 돌리면 비용이 두 배가 되지만, 1단계 필터로 70~80%를 먼저 걸러내면 추가 비용이 20~30% 수준으로 줄어듭니다.
다음 그림은 이 2단계 가드레일 시스템의 전체 흐름입니다. LLM이 생성한 답변이 어떤 경로를 거쳐 사용자에게 전달되는지 보여줍니다.
1단계 시맨틱 유사도 검사는 50~100ms로 빠르고 비용이 거의 없습니다. 70~80%의 답변이 여기서 바로 통과합니다. 나머지 20~30%의 의심스러운 답변만 2단계 LLM 심층 검증을 거치는데, 1~3초가 걸리지만 할루시네이션과 정책 위반을 정밀하게 잡아냅니다. 검증에 실패하면 안전한 대체 답변이 반환됩니다.
소규모 기업 환경에 맞게 구현합니다. DoorDash는 Claude 3 Haiku를 Amazon Bedrock에서 사용했는데, 로컬 Ollama 모델로 대체해도 같은 패턴을 적용할 수 있습니다.
# rag_generator.py """RAG 답변 생성기 + 가드레일 시스템.""" from dataclasses import dataclass import numpy as np from ollama_llm_client import OllamaLLMClient from embedding_service import OllamaEmbedding @dataclass class GeneratedAnswer: answer: str context_chunks: list[str] confidence: float # 가드레일 신뢰도 guardrail_passed: bool guardrail_details: dict class RAGGenerator: """검색된 컨텍스트 기반으로 답변을 생성하고 가드레일로 검증합니다.""" def __init__( self, llm_client: OllamaLLMClient, embedding_service: OllamaEmbedding, similarity_threshold: float = 0.7, use_llm_judge: bool = True, ): self.llm = llm_client self.embedding = embedding_service self.similarity_threshold = similarity_threshold self.use_llm_judge = use_llm_judge def generate(self, query: str, search_results: list) -> GeneratedAnswer: """검색 결과를 기반으로 답변을 생성하고 검증합니다.""" # 1. 컨텍스트 구성 context_chunks = [r.content for r in search_results] context = self._build_context(context_chunks) # 인덱싱 시점의 임베딩이 있으면 재활용 (Stage 1 성능 최적화) chunk_embeddings = [ r.embedding for r in search_results if hasattr(r, "embedding") and r.embedding is not None ] or None # 2. 답변 생성 answer = self._generate_answer(query, context) # 3. 가드레일 검증 guardrail_result = self._run_guardrails(answer, context_chunks, chunk_embeddings) return GeneratedAnswer( answer=answer if guardrail_result["passed"] else self._fallback_answer(query), context_chunks=context_chunks, confidence=guardrail_result["confidence"], guardrail_passed=guardrail_result["passed"], guardrail_details=guardrail_result, ) def _build_context(self, chunks: list[str]) -> str: """검색된 청크를 하나의 컨텍스트 문자열로 조합합니다.""" context_parts = [] for i, chunk in enumerate(chunks): context_parts.append(f"[문서 {i + 1}]\n{chunk}") return "\n\n".join(context_parts) def _generate_answer(self, query: str, context: str) -> str: """LLM으로 답변을 생성합니다.""" prompt = f"""아래 제공된 문서를 기반으로 질문에 답변하세요. 규칙: - 제공된 문서의 내용만을 기반으로 답변하세요. - 문서에 없는 정보는 추측하지 마세요. - 답변할 수 없는 경우 "제공된 문서에서 관련 정보를 찾을 수 없습니다"라고 답하세요. - 답변의 근거가 되는 문서 번호를 언급하세요. 문서: {context} 질문: {query} 답변:""" return self.llm.generate(prompt) def _run_guardrails( self, answer: str, context_chunks: list[str], chunk_embeddings: list[list[float]] = None ) -> dict: """2단계 가드레일을 실행합니다.""" # Stage 1: 시맨틱 유사도 검사 stage1_result = self._stage1_similarity_check(answer, context_chunks, chunk_embeddings) if stage1_result["score"] >= self.similarity_threshold: return { "passed": True, "confidence": stage1_result["score"], "stage": "stage1_pass", "details": stage1_result, } # Stage 2: LLM Judge (Stage 1 미통과 시) if self.use_llm_judge: stage2_result = self._stage2_llm_judge(answer, context_chunks) return { "passed": stage2_result["is_grounded"], "confidence": stage2_result["confidence"], "stage": "stage2_judge", "details": {**stage1_result, **stage2_result}, } # LLM Judge 비활성화 시 Stage 1 결과로 판단 return { "passed": False, "confidence": stage1_result["score"], "stage": "stage1_fail", "details": stage1_result, } def _stage1_similarity_check( self, answer: str, context_chunks: list[str], chunk_embeddings: list[list[float]] = None ) -> dict: """Stage 1: 답변과 컨텍스트의 시맨틱 유사도를 확인합니다. chunk_embeddings가 전달되면 인덱싱 시점의 임베딩을 재활용합니다. 없으면 직접 계산합니다(폴백). """ answer_embedding = self.embedding.embed(answer) max_similarity = 0.0 best_chunk_idx = -1 for i, chunk in enumerate(context_chunks): if chunk_embeddings and i < len(chunk_embeddings): chunk_embedding = chunk_embeddings[i] else: chunk_embedding = self.embedding.embed(chunk) similarity = self._cosine_similarity(answer_embedding, chunk_embedding) if similarity > max_similarity: max_similarity = similarity best_chunk_idx = i return { "score": max_similarity, "best_matching_chunk": best_chunk_idx, } def _stage2_llm_judge(self, answer: str, context_chunks: list[str]) -> dict: """Stage 2: LLM이 답변의 근거성을 판단합니다.""" context = "\n\n".join( f"[문서 {i + 1}]: {chunk}" for i, chunk in enumerate(context_chunks) ) judge_prompt = f"""당신은 RAG 시스템의 답변 품질 검증관입니다. 아래 답변이 제공된 문서에 근거하고 있는지 판단하세요. 판단 기준: 1. 근거성: 답변의 모든 주장이 제공된 문서에서 확인 가능한가? 2. 정확성: 문서의 내용을 왜곡하거나 과장하지 않았는가? 3. 완전성: 질문에 대해 충분히 답변했는가? 제공된 문서: {context} 검증할 답변: {answer} 다음 형식으로 판단하세요: 근거성: [예/아니오] 신뢰도: [0.0~1.0 사이 숫자] 이유: [판단 이유 한 줄]""" judge_response = self.llm.generate(judge_prompt) return self._parse_judge_response(judge_response) def _parse_judge_response(self, response: str) -> dict: """LLM Judge의 응답을 파싱합니다. Ollama 로컬 모델은 지시한 형식을 정확히 따르지 않는 경우가 잦으므로, 여러 긍정 표현을 체크하고 전체 응답에서 검색합니다. """ if not response: return {"is_grounded": False, "confidence": 0.0, "raw_response": ""} response_lower = response.lower() positive_markers = ["근거성: 예", "근거성: 네", "근거성: yes", "grounded: yes"] negative_markers = ["근거성: 아니오", "근거성: 아니요", "근거성: no", "grounded: no"] is_grounded = False for marker in positive_markers: if marker in response_lower: is_grounded = True break # 명시적 부정이 없고 긍정도 없으면, 전체 텍스트에서 판단 if not is_grounded and not any(m in response_lower for m in negative_markers): grounded_line = [l for l in response.split("\n") if "근거" in l] if grounded_line: is_grounded = any(w in grounded_line[0] for w in ["예", "네", "yes", "충분"]) confidence = 0.5 for line in response.split("\n"): if "신뢰도" in line or "confidence" in line.lower(): import re match = re.search(r"(\d+\.?\d*)", line) if match: val = float(match.group(1)) # 0~1 범위가 아니면 100 기준으로 변환 if val > 1.0: val = val / 100.0 confidence = min(max(val, 0.0), 1.0) break return { "is_grounded": is_grounded, "confidence": confidence, "raw_response": response, } def _fallback_answer(self, query: str) -> str: """가드레일 미통과 시 안전한 폴백 답변을 반환합니다.""" return ( "죄송합니다. 검색된 문서를 기반으로 신뢰할 수 있는 답변을 " "생성하지 못했습니다. 질문을 더 구체적으로 바꾸거나, " "관련 부서에 직접 문의해 주세요." ) @staticmethod def _cosine_similarity(a: list[float], b: list[float]) -> float: """두 벡터의 코사인 유사도를 계산합니다.""" a_arr = np.array(a) b_arr = np.array(b) dot = np.dot(a_arr, b_arr) norm = np.linalg.norm(a_arr) * np.linalg.norm(b_arr) if norm == 0: return 0.0 return float(dot / norm)
_run_guardrails 메서드의 흐름을 따라가 보면, DoorDash 패턴이 어떻게 작동하는지 보입니다. Stage 1에서 답변과 컨텍스트 청크 사이의 시맨틱 유사도를 계산합니다. 유사도가 0.7 이상이면 "이 답변은 컨텍스트에 근거하고 있을 가능성이 높다"고 판단해서 바로 통과시킵니다. 0.7 미만이면 Stage 2로 넘어가서 별도의 LLM이 답변의 근거성을 판단합니다.
여기서 성능 최적화가 중요합니다. DoorDash가 Stage 1을 "저비용"이라고 한 것은 임베딩이 인덱싱 시점에 이미 계산되어 있었기 때문입니다. 검색 결과에 임베딩 벡터가 포함되어 있으면 그것을 재활용하고, 새로 계산하는 것은 답변의 임베딩 하나뿐입니다. 청크마다 매번 embed()를 호출하면 Ollama 기준으로 청크 5개에 500ms~1초가 추가되므로, "저비용" Stage 1이 아니라 "중비용" Stage 1이 됩니다. 코드에서 chunk_embeddings 파라미터가 바로 이 최적화를 위한 것입니다.
Stage 2의 _stage2_llm_judge는 Ollama 로컬 모델로 돌립니다. 근거성, 정확성, 완전성 세 가지 기준으로 판단하고, "예/아니오"와 신뢰도 점수를 반환합니다. 로컬 모델이니 API 비용은 없지만, 추론 시간이 추가됩니다. llama3.2 기준으로 Stage 2 판단에 1~3초가 걸립니다.
가드레일을 통과하지 못한 답변은 _fallback_answer로 대체됩니다. 사용자에게 "답변을 신뢰하기 어려우니 다시 시도하거나 담당 부서에 문의하라"는 안내가 나갑니다. 틀린 답변을 자신있게 내보내는 것보다 "모르겠다"고 솔직히 말하는 것이 낫습니다. 특히 사내 시스템에서는 잘못된 정보가 업무 결정에 영향을 줄 수 있기 때문에 이 판단이 중요합니다.
similarity_threshold는 0.7이 기본값인데, 도메인에 따라 조정이 필요합니다. 기술 문서처럼 정확한 용어가 많은 도메인에서는 0.75로 높여도 됩니다. 자연어 대화형 질문이 많은 환경에서는 0.65로 낮추는 게 나을 수 있습니다. 이 임계값 튜닝도 뒤에서 다룰 RAGAS 평가와 연결됩니다. 평가 메트릭으로 최적 임계값을 찾을 수 있습니다.
DoorDash의 실제 결과를 보면, 이 2단계 구조로 할루시네이션이 90% 감소했고, 심각한 규정 위반이 99% 감소했습니다. 수십만 건의 일일 요청에 적용해서 2.5초 레이턴시를 유지한 것도 주목할 점입니다. 모든 요청에 Stage 2를 돌렸다면 레이턴시가 크게 늘었을 텐데, Stage 1에서 대부분을 걸러냄으로써 평균 레이턴시를 낮게 유지했습니다.
RAGAS로 품질을 측정하고 관리하기
검색 파이프라인을 만들었습니다. 가드레일도 달았습니다. 그런데 중요한 질문이 남아 있습니다. 이 시스템이 실제로 얼마나 잘 작동하는지, 어떻게 알 수 있을까요?
1편에서 Retrieval Decay 문제를 다뤘습니다. 문서가 늘어나면서 검색 품질이 조용히 하락하는 현상인데, 시스템은 에러 없이 돌아가고 사용자가 불만을 제기하기 전까지는 문제를 모릅니다. "바이브 체크"로는 이 문제를 못 잡습니다. 정기적인 평가 루프가 필요합니다.
RAGAS는 참조 답변 없이 RAG 파이프라인을 평가하는 프레임워크입니다. GitHub 스타 8.7K로, RAG 평가에서는 사실상 표준으로 자리 잡았습니다. 수동으로 레이블링한 정답 데이터셋 없이도 LLM-as-judge 방식으로 평가할 수 있어서 소규모 기업에 특히 쓸 만합니다.
RAGAS의 핵심 메트릭 세 가지를 살펴보겠습니다.
첫째, Context Precision입니다. 검색된 컨텍스트 청크 중에서 실제로 답변에 사용된 비율을 측정합니다. 10개 청크를 검색했는데 답변에 실제로 도움이 된 것이 3개뿐이라면 Context Precision은 0.3입니다. 이 수치가 낮으면 검색 결과에 관련 없는 정보가 너무 많이 섞여 있다는 뜻입니다. 하이브리드 검색의 weight 조정이나 리랭킹의 top_k 조절이 필요합니다.
둘째, Context Recall입니다. 정확한 답변에 필요한 모든 정보가 검색된 컨텍스트에 있는지 평가합니다. 답변에 필요한 핵심 정보가 검색 결과에 빠져 있으면 Context Recall이 낮아집니다. 이 수치가 낮으면 검색 범위를 넓히거나 (initial_k를 늘리거나), 청킹 전략을 조정해야 합니다.
셋째, Faithfulness입니다. 생성된 답변이 검색된 컨텍스트에 충실한지 측정합니다. 답변에 포함된 주장 하나하나의 근거를 컨텍스트에서 찾을 수 있는지 확인합니다. Faithfulness가 낮으면 LLM이 컨텍스트에 없는 내용을 지어내고 있다는 뜻입니다. 앞에서 만든 가드레일의 effectiveness를 측정하는 지표이기도 합니다.
코드로 구현합니다. RAGAS 라이브러리를 사용하되, 평가용 LLM도 Ollama 로컬 모델로 대체합니다.
# evaluator.py """RAGAS 기반 RAG 평가 시스템.""" import json from datetime import datetime from dataclasses import dataclass, asdict from ragas import evaluate from ragas.metrics import ( context_precision, context_recall, faithfulness, answer_relevancy, ) from ragas.llms import LangchainLLMWrapper from ragas.embeddings import LangchainEmbeddingsWrapper from langchain_ollama import ChatOllama, OllamaEmbeddings # pip install langchain-ollama from datasets import Dataset @dataclass class EvalCase: """평가 케이스 하나를 정의합니다.""" question: str ground_truth: str # 기대 답변 (없으면 빈 문자열) contexts: list[str] # 검색된 컨텍스트 answer: str # 시스템이 생성한 답변 @dataclass class EvalResult: """평가 결과.""" timestamp: str num_cases: int metrics: dict[str, float] per_case_scores: list[dict] class RAGEvaluator: """RAGAS를 사용한 RAG 평가기.""" def __init__( self, llm_model: str = "llama3.2", embedding_model: str = "nomic-embed-text", ): # RAGAS에서 사용할 LLM과 임베딩을 Ollama로 설정 llm = ChatOllama(model=llm_model) embeddings = OllamaEmbeddings(model=embedding_model) self.llm = LangchainLLMWrapper(llm) self.embeddings = LangchainEmbeddingsWrapper(embeddings) self.metrics = [ context_precision, context_recall, faithfulness, answer_relevancy, ] def evaluate(self, eval_cases: list[EvalCase]) -> EvalResult: """평가 케이스 목록을 RAGAS로 평가합니다.""" # RAGAS Dataset 형식으로 변환 data = { "question": [c.question for c in eval_cases], "ground_truth": [c.ground_truth for c in eval_cases], "contexts": [c.contexts for c in eval_cases], "answer": [c.answer for c in eval_cases], } dataset = Dataset.from_dict(data) # 평가 실행 result = evaluate( dataset, metrics=self.metrics, llm=self.llm, embeddings=self.embeddings, ) # 결과 정리 df = result.to_pandas() metrics_avg = { "context_precision": float(df["context_precision"].mean()), "context_recall": float(df["context_recall"].mean()), "faithfulness": float(df["faithfulness"].mean()), "answer_relevancy": float(df["answer_relevancy"].mean()), } per_case = df.to_dict(orient="records") return EvalResult( timestamp=datetime.now().isoformat(), num_cases=len(eval_cases), metrics=metrics_avg, per_case_scores=per_case, ) def save_result(self, result: EvalResult, path: str): """평가 결과를 JSON 파일로 저장합니다.""" with open(path, "w", encoding="utf-8") as f: json.dump(asdict(result), f, ensure_ascii=False, indent=2) print(f"평가 결과 저장: {path}") print(f" Context Precision: {result.metrics['context_precision']:.3f}") print(f" Context Recall: {result.metrics['context_recall']:.3f}") print(f" Faithfulness: {result.metrics['faithfulness']:.3f}") print(f" Answer Relevancy: {result.metrics['answer_relevancy']:.3f}")
여기서 핵심은 평가 데이터셋을 어떻게 만드느냐입니다. 처음부터 완벽한 데이터셋을 만들 필요는 없습니다. 실무에서 권장하는 방법은 이렇습니다.
# eval_dataset_builder.py """평가 데이터셋 생성기.""" from evaluator import EvalCase def build_initial_eval_set() -> list[EvalCase]: """초기 평가 데이터셋을 수동으로 만듭니다. 20~30개의 대표 질문을 선정합니다. - 팩트 조회형 질문 10개 (정확한 답이 하나인 경우) - 설명형 질문 10개 (맥락이 필요한 경우) - 엣지 케이스 질문 5~10개 (답변이 문서에 없는 경우) """ cases = [ # 팩트 조회형 EvalCase( question="휴가 신청은 어떻게 하나요?", ground_truth="인사 포털에서 휴가 신청 메뉴를 통해 신청합니다.", contexts=[], # 평가 시 실제 검색 결과로 채워짐 answer="", # 평가 시 실제 생성 답변으로 채워짐 ), EvalCase( question="회사 VPN 접속 설정 방법을 알려주세요.", ground_truth="IT 지원팀에서 제공하는 VPN 클라이언트를 설치 후 사번으로 로그인합니다.", contexts=[], answer="", ), # 설명형 EvalCase( question="신규 입사자 온보딩 절차는 어떻게 되나요?", ground_truth="입사일 기준 1주차 OT, 2주차 부서 배치, 3주차 멘토링이 진행됩니다.", contexts=[], answer="", ), # 엣지 케이스 (문서에 없는 질문) EvalCase( question="사무실에서 고양이를 키울 수 있나요?", ground_truth="관련 정책을 찾을 수 없습니다.", contexts=[], answer="", ), ] return cases def run_eval_with_pipeline(pipeline, eval_cases: list[EvalCase]) -> list[EvalCase]: """실제 파이프라인을 돌려서 평가 데이터를 채웁니다.""" filled_cases = [] for case in eval_cases: # 검색 search_results = pipeline.search(case.question) contexts = [r.content for r in search_results] # 답변 생성 generated = pipeline.generate(case.question, search_results) filled_cases.append(EvalCase( question=case.question, ground_truth=case.ground_truth, contexts=contexts, answer=generated.answer, )) return filled_cases
처음에 20~30개의 대표 질문을 수동으로 만들고, 실제 시스템을 돌려서 검색 결과와 답변을 채운 뒤 RAGAS로 평가합니다. 이 과정을 매주 또는 월간으로 반복하면 시스템 품질의 추이를 추적할 수 있습니다. 특히 새 문서를 대량으로 추가한 뒤에는 반드시 평가를 돌려서 Retrieval Decay가 발생하지 않았는지 확인해야 합니다.
Cohorte의 RAGAS 심층 분석에서 제시하는 메트릭별 해석 기준이 실무에서 유용합니다. Context Precision이 낮으면 검색이 너무 많은 무관 문서를 가져오고 있다는 의미이므로 리랭킹의 top_k를 줄이거나 similarity threshold를 높여야 합니다. Context Recall이 낮으면 필요한 정보가 검색 결과에서 빠지고 있으므로 initial_k를 늘리거나 BM25 가중치를 조정해야 합니다. Faithfulness가 낮으면 LLM이 컨텍스트에 없는 내용을 만들어내고 있으므로 프롬프트를 강화하거나 가드레일 임계값을 높여야 합니다.
RAGAS 설치는 다음과 같습니다.
pip install ragas datasets langchain-ollama
RAGAS와 LangChain은 패키지 구조가 자주 바뀝니다. 위 코드는 ragas 0.2.x, langchain-ollama 0.3.x 기준입니다. 버전에 따라 import 경로가 다를 수 있으니 각 패키지의 공식 문서를 확인하세요.
관측성 - Langfuse로 프로덕션 운영하기
RAGAS는 오프라인 평가입니다. 주기적으로 평가 데이터셋을 돌려서 점수를 확인하는 방식입니다. 프로덕션에서는 실시간 모니터링도 필요합니다. 사용자 쿼리마다 검색 품질, 응답 시간, 가드레일 트리거 여부를 추적해야 합니다. 문제가 생겼을 때 "어느 단계에서 무엇이 잘못됐는지"를 바로 확인할 수 있어야 합니다.
Langfuse가 이 목적에 잘 맞습니다. GitHub 스타 19K의 오픈소스 관측성 플랫폼으로, 셀프 호스팅이 됩니다. 트레이싱, 프롬프트 버전 관리, 평가 점수 기록을 지원하고요. Langfuse 블로그의 RAG 관측성 가이드에서도 소규모 팀(3~10명)에게 범용 솔루션으로 Langfuse를 추천합니다.
Langfuse를 RAG 파이프라인에 통합하는 코드입니다.
# observability.py """Langfuse 관측성 통합.""" import time from functools import wraps from langfuse import Langfuse from langfuse.decorators import observe, langfuse_context # Langfuse 초기화 (환경 변수 또는 직접 설정) # LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_HOST를 환경 변수로 설정 langfuse = Langfuse() class ObservableRAGPipeline: """Langfuse 트레이싱이 통합된 RAG 파이프라인.""" def __init__(self, hybrid_retriever, reranker, generator): self.retriever = hybrid_retriever self.reranker = reranker self.generator = generator @observe(name="rag-query") def query(self, question: str, user_id: str = None) -> dict: """사용자 질문을 처리합니다. 전체 흐름이 Langfuse에 트레이싱됩니다.""" # 1. 검색 search_start = time.time() search_results = self._search(question) search_time = time.time() - search_start # 2. 리랭킹 rerank_start = time.time() ranked_results = self._rerank(question, search_results) rerank_time = time.time() - rerank_start # 3. 답변 생성 + 가드레일 gen_start = time.time() answer = self._generate(question, ranked_results) gen_time = time.time() - gen_start # 메타데이터 기록 langfuse_context.update_current_trace( user_id=user_id, metadata={ "search_time_ms": round(search_time * 1000), "rerank_time_ms": round(rerank_time * 1000), "generation_time_ms": round(gen_time * 1000), "total_time_ms": round((search_time + rerank_time + gen_time) * 1000), "num_search_results": len(search_results), "num_reranked": len(ranked_results), "guardrail_passed": answer.guardrail_passed, "confidence": answer.confidence, }, ) # 가드레일 실패 시 점수 기록 if not answer.guardrail_passed: langfuse_context.score_current_trace( name="guardrail_triggered", value=0, comment="가드레일 미통과로 폴백 답변 반환", ) return { "answer": answer.answer, "confidence": answer.confidence, "guardrail_passed": answer.guardrail_passed, "sources": [r.doc_id for r in ranked_results], } @observe(name="hybrid-search") def _search(self, question: str): return self.retriever.search(question, top_k=10, initial_k=30) @observe(name="rerank") def _rerank(self, question: str, results): return self.reranker.rerank(question, results, top_k=5) @observe(name="generate-with-guardrail") def _generate(self, question: str, results): return self.generator.generate(question, results)
@observe 데코레이터만 붙이면 각 메서드의 입력, 출력, 실행 시간이 자동으로 Langfuse에 기록됩니다. 각 사용자 요청은 하나의 trace로 묶이고, 그 안에서 검색, 리랭킹, 생성 단계가 개별 span으로 기록됩니다.
Langfuse 셀프 호스팅은 Docker Compose로 간단히 올릴 수 있습니다.
# Langfuse 셀프 호스팅 (Docker Compose) git clone https://github.com/langfuse/langfuse.git cd langfuse docker compose up -d
셀프 호스팅이 부담스러우면 Langfuse Cloud의 무료 플랜(월 50K events)으로 시작해도 됩니다. 소규모 기업의 초기 트래픽이면 무료 플랜으로 충분합니다.
Langfuse 대시보드에서는 쿼리별 전체 처리 시간과 각 단계의 소요 시간 분포를 볼 수 있습니다. 가드레일 트리거 비율도 추적되는데, 이 비율이 갑자기 올라가면 검색 품질에 문제가 생겼다는 신호입니다. 특정 쿼리가 반복적으로 낮은 confidence를 보이면 해당 도메인의 문서가 부족하다는 뜻이고요.
RAGAS 오프라인 평가와 Langfuse 실시간 모니터링을 결합하면 품질 관리 체계가 완성됩니다. RAGAS는 주간/월간 정기 평가로 전체적인 품질 추세를 파악하고, Langfuse는 실시간으로 개별 쿼리의 문제를 감지합니다.
전체 검색 파이프라인 조립
지금까지 만든 구성 요소를 하나의 파이프라인으로 조립합니다. 2편의 데이터 수집 파이프라인과 이번 편의 검색 파이프라인을 합쳐서, 문서 인덱싱부터 질의응답까지 하나의 시스템으로 만듭니다.
# search_pipeline.py """RAG 검색 파이프라인. 2편의 데이터 수집 파이프라인(pipeline.py) 위에 검색, 리랭킹, 생성, 가드레일, 평가를 통합합니다. """ import json from pathlib import Path from bm25_retriever import BM25Retriever, BM25Document from hybrid_retriever import HybridRetriever from reranker import Reranker from rag_generator import RAGGenerator from embedding_service import OllamaEmbedding from vector_store import VectorStore from ollama_llm_client import OllamaLLMClient class SearchPipeline: """데이터 검색부터 답변 생성까지의 전체 파이프라인.""" def __init__( self, persist_dir: str = "./chroma_db", embedding_model: str = "nomic-embed-text", llm_model: str = "llama3.2", rerank_model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2", similarity_threshold: float = 0.7, ): # 기반 서비스 초기화 self.embedding = OllamaEmbedding(model=embedding_model) self.llm = OllamaLLMClient(model=llm_model) self.vector_store = VectorStore(persist_dir=persist_dir) # BM25 인덱스 구축 self.bm25 = BM25Retriever() self._build_bm25_index() # 하이브리드 검색기 self.hybrid_retriever = HybridRetriever( vector_store=self.vector_store, embedding_service=self.embedding, bm25_retriever=self.bm25, ) # 리랭커 self.reranker = Reranker(model_name=rerank_model) # 답변 생성기 + 가드레일 self.generator = RAGGenerator( llm_client=self.llm, embedding_service=self.embedding, similarity_threshold=similarity_threshold, ) def _build_bm25_index(self): """벡터 저장소의 문서로 BM25 인덱스를 구축합니다.""" # ChromaDB에서 모든 문서를 가져와 BM25 인덱스에 추가 all_docs = self.vector_store.get_all_documents() bm25_docs = [ BM25Document( doc_id=doc["id"], content=doc["content"], metadata=doc.get("metadata", {}), ) for doc in all_docs ] self.bm25.index(bm25_docs) print(f"BM25 인덱스 구축 완료: {len(bm25_docs)}개 문서") def query(self, question: str) -> dict: """사용자 질문에 답변합니다. 전체 흐름: 1. 하이브리드 검색 (벡터 + BM25 + RRF) 2. 크로스인코더 리랭킹 3. LLM 답변 생성 4. 2단계 가드레일 검증 """ # 1. 하이브리드 검색 search_results = self.hybrid_retriever.search( question, top_k=10, initial_k=30 ) if not search_results: return { "answer": "검색 결과가 없습니다. 질문을 다르게 표현해 보세요.", "confidence": 0.0, "sources": [], } # 2. 리랭킹 ranked_results = self.reranker.rerank( question, search_results, top_k=5 ) # 3. 답변 생성 + 가드레일 generated = self.generator.generate(question, ranked_results) return { "answer": generated.answer, "confidence": generated.confidence, "guardrail_passed": generated.guardrail_passed, "sources": [ { "doc_id": r.doc_id, "score": r.rerank_score, "sources": r.sources, } for r in ranked_results ], } def main(): """대화형 RAG 시스템.""" import argparse parser = argparse.ArgumentParser(description="RAG 검색 파이프라인") parser.add_argument("--persist-dir", default="./chroma_db", help="벡터 DB 경로") parser.add_argument("--embedding-model", default="nomic-embed-text") parser.add_argument("--llm-model", default="llama3.2") args = parser.parse_args() print("RAG 검색 파이프라인을 초기화합니다...") pipeline = SearchPipeline( persist_dir=args.persist_dir, embedding_model=args.embedding_model, llm_model=args.llm_model, ) print("초기화 완료. 질문을 입력하세요. (종료: quit)\n") while True: question = input("질문> ").strip() if question.lower() in ("quit", "exit", "q"): break if not question: continue result = pipeline.query(question) print(f"\n답변: {result['answer']}") print(f"신뢰도: {result['confidence']:.2f}") if not result.get("guardrail_passed", True): print("(가드레일에 의해 폴백 답변이 반환되었습니다)") print(f"참조 문서: {len(result['sources'])}개") print() if __name__ == "__main__": main()
이 파이프라인을 실행하면 대화형 RAG 시스템이 됩니다.
# 검색 파이프라인 실행 python search_pipeline.py --persist-dir ./chroma_db # 출력 예시: # RAG 검색 파이프라인을 초기화합니다... # BM25 인덱스 구축 완료: 1,247개 문서 # 초기화 완료. 질문을 입력하세요. (종료: quit) # # 질문> 휴가 신청 절차를 알려주세요 # 답변: 인사 포털(hr.company.com)에 접속하여 ... # 신뢰도: 0.87 # 참조 문서: 5개
2편에서 pipeline.py --input-dir ./documents로 인덱싱하고, 이번 편의 search_pipeline.py로 검색하는 구조입니다. 두 스크립트가 같은 ./chroma_db 디렉토리를 공유합니다.
전체 시스템의 설치와 실행을 정리합니다.
# 필요한 패키지 전체 설치 pip install chromadb langchain langchain-ollama \ sentence-transformers torch \ ragas datasets \ langfuse \ kiwipiepy \ numpy pymupdf beautifulsoup4 requests # Ollama 모델 다운로드 ollama pull nomic-embed-text # 임베딩 ollama pull llama3.2 # LLM # 1. 문서 인덱싱 (2편의 파이프라인) python pipeline.py --input-dir ./documents --persist-dir ./chroma_db # 2. 대화형 검색 (이번 편의 파이프라인) python search_pipeline.py --persist-dir ./chroma_db # 3. 평가 실행 python -c " from evaluator import RAGEvaluator from eval_dataset_builder import build_initial_eval_set, run_eval_with_pipeline from search_pipeline import SearchPipeline pipeline = SearchPipeline() cases = build_initial_eval_set() filled = run_eval_with_pipeline(pipeline, cases) evaluator = RAGEvaluator() result = evaluator.evaluate(filled) evaluator.save_result(result, 'eval_results.json') "
각 단계별 예상 소요 시간을 정리합니다. M1 MacBook, CPU 모드, 문서 1,000개 기준 측정값입니다.
| 단계 | 소요 시간 | 비고 |
|---|---|---|
| 벡터 검색 (ChromaDB) | 50~100ms | 문서 수에 비례하지만 1만 개까지 100ms 이내 |
| BM25 검색 | 10~30ms | 인메모리 검색이므로 빠름 |
| RRF 병합 | <5ms | 단순 점수 계산 |
| 크로스인코더 리랭킹 | 200~500ms | 10개 문서 기준, GPU 사용 시 50ms 이내 |
| LLM 답변 생성 (llama3.2) | 2~5초 | 가장 오래 걸리는 단계, GPU 시 1~2초 |
| 가드레일 Stage 1 | 50~100ms | 임베딩 재활용 시, 미재활용 시 500ms~1초 |
| 가드레일 Stage 2 (트리거 시) | 1~3초 | LLM 추가 호출, 전체 요청의 20~30%에서 발생 |
Stage 2가 트리거되지 않는 일반적인 경우 전체 응답 시간은 3~6초, Stage 2까지 실행되면 5~9초입니다. GPU가 있으면 전체적으로 2~4초로 줄어듭니다. 사내 QA 시스템으로는 충분히 실용적인 수준입니다.
이 구성의 월간 운영 비용을 정리합니다. Ollama를 로컬에서 실행하므로 임베딩과 LLM 추론 비용이 없습니다. ms-marco-MiniLM 리랭킹 모델도 로컬 실행입니다. Langfuse 셀프 호스팅은 기존 서버에서 Docker 컨테이너로 돌리면 추가 비용이 없습니다. ChromaDB도 로컬 저장소입니다. 기존 서버나 개발 장비를 활용하면 추가 비용이 사실상 없습니다. GPU가 없어도 CPU 모드로 모든 구성 요소가 작동합니다. GPU가 있으면 리랭킹과 LLM 추론이 빨라지지만, 소규모 기업의 쿼리 볼륨에서는 CPU로도 충분히 실용적인 응답 시간(3~8초)을 기대할 수 있습니다.
마무리
돌아보면, 이번 편은 "검색된 결과를 어떻게 믿을 수 있게 만드느냐"에 대한 이야기였습니다.
벡터 검색에 BM25를 얹고 RRF로 합치는 하이브리드 검색이 출발점이었고, 크로스인코더 리랭킹으로 정밀도를 한 단계 더 끌어올렸습니다. HyDE 쿼리 변환은 짧은 질문에 대한 보험이고요. 수치로 보면 벡터 검색 단독 대비 40~50% 정확도가 올라가는 셈입니다.
답변의 신뢰성은 DoorDash의 2단계 가드레일 패턴으로 잡았습니다. 시맨틱 유사도로 빠르게 거르고, 의심스러운 것만 LLM Judge로 정밀 검증하는 구조인데, 비용 효율성과 정확성을 동시에 잡는 방법입니다.
그리고 이 모든 것이 제대로 돌아가는지 측정하는 체계도 갖췄습니다. RAGAS로 오프라인 평가, Langfuse로 실시간 모니터링. search_pipeline.py 한 파일에 전부 들어가고, Ollama 로컬 실행이라 추가 비용도 없습니다.
개발자 커뮤니티에서 자주 받는 질문 중 하나가 "이걸 전부 다 해야 하나요?"인데, 솔직히 말하면 순서가 중요합니다. 하이브리드 검색이 1순위입니다. 벡터 검색에 BM25를 추가하는 것만으로도 체감 품질이 확 달라지고, 1~2주면 구현 가능하거든요. 그 다음이 가드레일입니다. 사내 시스템에서 잘못된 답변이 나가면 심각한 문제가 되니, 최소한 시맨틱 유사도 기반의 1단계 가드레일은 처음부터 넣으세요. 리랭킹은 검색 품질이 일정 수준에 올라온 뒤에 추가해도 되고, RAGAS 평가와 Langfuse 모니터링은 프로덕션에 올린 뒤 점진적으로 도입하면 됩니다.
다음 편은 이 시리즈의 마지막입니다. 지금까지 만든 RAG 시스템을 MCP 서버로 포장해서, Claude가 회사의 지식을 직접 검색하고 활용할 수 있게 만듭니다. Claude Desktop이나 Claude Code에 연결하면, "우리 회사 휴가 규정이 뭐야?"라는 질문에 Claude가 사내 문서를 직접 뒤져서 답하는 식이죠. FastMCP 3.0으로 구현하고, 인증과 권한 관리까지 다룹니다.
참고 자료
- Optimizing RAG with Hybrid Search and Reranking - Superlinked VectorHub
- Choosing the Best Reranking Model (2026 Guide) - Zero Entropy
- Hybrid Search vs Reranker in RAG - bswen
- Enterprise RAG Architecture: A Practitioner's Guide - Applied AI
- Building a High-Quality RAG-Based Support System (DoorDash) - ZenML
- RAGAS Documentation - RAGAS 공식 문서
- Evaluating RAG Systems in 2025: RAGAS Deep Dive - Cohorte
- RAG Observability and Evals - Langfuse 블로그
- Six Lessons Learned Building RAG Systems in Production - Towards Data Science
- LangChain vs LlamaIndex: Production RAG in 2026
다음 편: [4편] MCP로 완성하기 - Claude가 회사 지식을 쓰게 만들기
댓글
댓글을 작성하려면 이 필요합니다.