본문 바로가기

RAG

RAG 데모는 잘 되는데 배포하면 망하는 이유 7가지 — 원인별 해결법, 프로덕션 RAG 완전 가이드

반응형

RAG 데모는 항상 잘 돼요.

PDF 몇 개 넣고, 벡터 DB 연결하고, LLM 붙이면 마법처럼 답이 나와요. 팀이 흥분하고, 경영진이 빠른 배포를 요구해요.

그리고 3개월 뒤, 시스템이 무너지기 시작해요.

데이터가 많아지면서 검색이 틀리고, 답이 엉뚱하고, 비용이 폭증하고, 아무도 원인을 모르는 상황이 돼요.

엔터프라이즈 RAG 구현의 40~72%가 첫 해 안에 실패해요. 모델이 나빠서가 아니에요. 아키텍처가 데모용으로 설계됐기 때문이에요.

이번 글에서는 RAG가 실패하는 7가지 이유와 실전 해결책을 코드와 함께 정리해 드릴게요.


실패 원인 1: 잘못된 청킹 전략

가장 흔하고 가장 치명적인 실수예요. 문서를 고정 크기(512 토큰)로 자르면 이런 일이 생겨요.

원본 문서:
"환불 정책은 구매 후 30일 이내에 적용됩니다.
단, 할인 상품의 경우 7일로 단축됩니다."

나쁜 청킹 (512 토큰 고정):
청크 1: "...환불 정책은 구매 후 30일 이내에"
청크 2: "적용됩니다. 단, 할인 상품의 경우 7일..."

→ "할인 상품 환불 기간은?" 질문에
→ 청크 2만 검색됨
→ "7일로 단축됩니다" 라는 불완전한 답변 반환

문장이 잘리고, 표가 깨지고, 문맥이 소실돼요.

해결책: 의미 기반 청킹

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# 나쁜 방법 — 고정 크기 청킹
bad_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=0  # 오버랩도 없음
)

# 좋은 방법 1 — 오버랩 포함 청킹
good_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=100,      # 앞뒤 문맥 유지
    separators=["\n\n", "\n", ".", "!", "?", " "]  # 의미 단위 기준
)

# 좋은 방법 2 — 시맨틱 청킹 (의미가 바뀔 때 자름)
semantic_splitter = SemanticChunker(
    embeddings=OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95  # 의미 유사도가 크게 떨어질 때 청크 분리
)

# 좋은 방법 3 — 계층적 청킹 (부모-자식 구조)
class HierarchicalChunker:
    def chunk(self, document: str) -> list[dict]:
        # 큰 청크 (문맥 저장용)
        parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2048, chunk_overlap=200)
        # 작은 청크 (검색용)
        child_splitter = RecursiveCharacterTextSplitter(chunk_size=256, chunk_overlap=50)

        parents = parent_splitter.split_text(document)
        chunks = []

        for parent_id, parent in enumerate(parents):
            children = child_splitter.split_text(parent)
            for child in children:
                chunks.append({
                    "content": child,          # 검색에 사용
                    "parent_content": parent,  # LLM에 전달 (더 넓은 문맥)
                    "parent_id": parent_id
                })

        return chunks

청킹 전략 선택 기준

문서 유형 권장 청크 크기 오버랩

기술 문서, API 레퍼런스 256~512 토큰 50~100
계약서, 법률 문서 512~1024 토큰 100~200
FAQ, 구조화된 문서 문단 단위 -
코드 함수/클래스 단위 -

실패 원인 2: 순수 벡터 검색의 한계

벡터 검색은 의미적 유사도를 찾는 데 강해요. 근데 정확한 키워드가 필요한 경우에 실패해요.

사용자 질문: "ISO 27001 컴플라이언스 요구사항 알려줘"

벡터 검색 결과:
1. "보안 모범 사례" (의미 유사, 키워드 없음)
2. "컴플라이언스 프레임워크 개요" (의미 유사, 키워드 없음)
3. "ISO 27001 구체적 요구사항" ← 진짜 필요한 문서
   (벡터 유사도 낮아서 3위로 밀림)

정작 필요한 문서가 뒤로 밀려요.

해결책: 하이브리드 검색 (BM25 + 벡터)

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

def create_hybrid_retriever(documents: list, vectorstore: Qdrant):
    # BM25 — 키워드 기반 검색
    bm25_retriever = BM25Retriever.from_documents(documents)
    bm25_retriever.k = 5

    # 벡터 검색 — 의미 기반 검색
    vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

    # 두 방식 앙상블 (가중치: 벡터 60% + BM25 40%)
    hybrid_retriever = EnsembleRetriever(
        retrievers=[vector_retriever, bm25_retriever],
        weights=[0.6, 0.4]
    )

    return hybrid_retriever

# 사용
retriever = create_hybrid_retriever(documents, vectorstore)
results = retriever.invoke("ISO 27001 컴플라이언스 요구사항")

하이브리드 검색은 시맨틱 검색만 쓸 때보다 평균 15~20% 검색 정확도가 높아요.


실패 원인 3: 리랭킹(Reranking) 없음

검색 결과 상위 5개를 그대로 LLM에 넘기는 건 위험해요. 벡터 검색이 뽑은 순서가 실제 관련도 순서와 다를 수 있어요.

해결책: Cross-Encoder 리랭킹

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

def create_reranking_retriever(base_retriever):
    # Cohere Rerank v3.5 사용 (또는 BGE-Reranker 셀프호스팅)
    compressor = CohereRerank(
        model="rerank-multilingual-v3.0",
        top_n=3  # 상위 3개만 남김
    )

    # 기본 검색 → 리랭킹 파이프라인
    compression_retriever = ContextualCompressionRetriever(
        base_compressor=compressor,
        base_retriever=base_retriever
    )

    return compression_retriever


# 비용 없이 셀프호스팅하고 싶다면 BGE-Reranker
from sentence_transformers import CrossEncoder

class LocalReranker:
    def __init__(self):
        self.model = CrossEncoder("BAAI/bge-reranker-v2-m3")

    def rerank(self, query: str, documents: list[str], top_k: int = 3) -> list[str]:
        pairs = [(query, doc) for doc in documents]
        scores = self.model.predict(pairs)

        # 점수 기준 정렬 후 상위 k개 반환
        ranked = sorted(zip(scores, documents), reverse=True)
        return [doc for _, doc in ranked[:top_k]]

리랭킹을 추가하면 RAG 정확도가 평균 10~25% 향상돼요. ~50ms 레이턴시 추가 비용은 품질 향상 대비 충분히 가치 있어요.


실패 원인 4: 쿼리 그대로 검색

사용자가 입력한 질문을 그대로 검색에 쓰면 두 가지 문제가 생겨요.

문제 1 — 모호한 쿼리

사용자: "그거 어떻게 해?"
→ "그거"가 뭔지 벡터 DB가 모름

문제 2 — 복잡한 쿼리

사용자: "A 기능과 B 기능의 차이점을 비교하고 어느 게 더 빠른지 알려줘"
→ 하나의 쿼리로는 두 개의 다른 정보가 필요

해결책: 쿼리 변환

from langchain.prompts import ChatPromptTemplate
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6")

# 1. 쿼리 재작성 (모호한 쿼리 명확화)
query_rewrite_prompt = ChatPromptTemplate.from_template("""
다음 사용자 질문을 문서 검색에 최적화된 형태로 재작성하세요.
전문 용어를 사용하고, 모호한 표현을 명확하게 바꾸세요.

원본 질문: {question}

재작성된 질문 (1개만):
""")

# 2. 다중 쿼리 생성 (복잡한 쿼리 분해)
multi_query_prompt = ChatPromptTemplate.from_template("""
다음 질문에 완전히 답하기 위해 필요한 검색 쿼리를 3개 생성하세요.
각 쿼리는 서로 다른 측면을 다뤄야 합니다.

원본 질문: {question}

검색 쿼리 (번호 없이, 한 줄씩):
""")

class QueryTransformer:
    def rewrite(self, query: str) -> str:
        chain = query_rewrite_prompt | llm
        return chain.invoke({"question": query}).content

    def decompose(self, query: str) -> list[str]:
        chain = multi_query_prompt | llm
        result = chain.invoke({"question": query}).content
        return [q.strip() for q in result.strip().split("\n") if q.strip()]

    def search_with_transform(self, query: str, retriever) -> list:
        # 복잡한 쿼리는 분해해서 검색
        sub_queries = self.decompose(query)
        all_results = []

        for sub_query in sub_queries:
            results = retriever.invoke(sub_query)
            all_results.extend(results)

        # 중복 제거
        seen = set()
        unique_results = []
        for doc in all_results:
            if doc.page_content not in seen:
                seen.add(doc.page_content)
                unique_results.append(doc)

        return unique_results

실패 원인 5: 데이터 신선도 관리 부재

RAG 시스템 실패의 1위 원인이에요. 문서가 업데이트돼도 인덱스가 업데이트되지 않으면 오래된 정보로 답해요.

예시:
- 2월: 환불 정책이 30일 → 14일로 변경됨
- 3월: 사용자가 "환불 기간이 얼마예요?" 질문
- 시스템: "30일입니다" 라고 답변 (구 정책)
→ 고객 불만, 법적 책임 가능성

해결책: 증분 인덱싱 파이프라인

import hashlib
from datetime import datetime
from supabase import create_client

supabase = create_client(SUPABASE_URL, SUPABASE_KEY)

class IncrementalIndexer:
    def get_content_hash(self, content: str) -> str:
        return hashlib.sha256(content.encode()).hexdigest()

    def should_reindex(self, doc_id: str, content: str) -> bool:
        """문서가 변경됐는지 확인"""
        result = supabase.table("document_hashes") \
            .select("hash") \
            .eq("doc_id", doc_id) \
            .execute()

        current_hash = self.get_content_hash(content)

        if not result.data:
            return True  # 새 문서

        return result.data[0]["hash"] != current_hash  # 변경됨

    def index_document(self, doc_id: str, content: str, vectorstore):
        if not self.should_reindex(doc_id, content):
            return  # 변경 없으면 스킵

        # 기존 청크 삭제
        vectorstore.delete(filter={"doc_id": doc_id})

        # 새로 청킹 및 인덱싱
        chunks = chunker.chunk(content)
        for chunk in chunks:
            chunk["metadata"]["doc_id"] = doc_id
        vectorstore.add_documents(chunks)

        # 해시 업데이트
        supabase.table("document_hashes").upsert({
            "doc_id": doc_id,
            "hash": self.get_content_hash(content),
            "updated_at": datetime.now().isoformat()
        }).execute()

    def run_scheduled_reindex(self, document_sources: list):
        """스케줄러로 주기적 실행 (cron, n8n 등)"""
        for source in document_sources:
            content = fetch_document(source["url"])
            self.index_document(source["id"], content, vectorstore)

실패 원인 6: 검색 실패를 탐지 못함

RAG가 잘못된 문서를 검색해도 LLM이 그럴듯하게 답하기 때문에 실패를 모르고 넘어가요.

실제 발생하는 일:
1. 사용자: "프리미엄 플랜 가격은?"
2. 검색: "베이직 플랜 가격표" 청크를 반환 (유사도 높음)
3. LLM: "베이직 플랜은 $10/월입니다" 라고 자신 있게 답변
4. 사용자: 틀린 정보를 믿음
→ 시스템은 이 실패를 모름

해결책: 검색 품질 평가 + Groundedness 체크

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6")

class RAGQualityChecker:

    def check_retrieval_relevance(
        self,
        query: str,
        retrieved_docs: list[str]
    ) -> dict:
        """검색된 문서가 쿼리와 관련 있는지 평가"""
        prompt = f"""
        사용자 질문: {query}

        검색된 문서들:
        {chr(10).join([f"{i+1}. {doc[:200]}..." for i, doc in enumerate(retrieved_docs)])}

        이 문서들이 질문에 답하기에 충분한 정보를 담고 있나요?
        JSON으로만 답하세요:
        {{"is_relevant": true/false, "confidence": 0.0~1.0, "reason": "이유"}}
        """

        import json
        result = json.loads(llm.invoke(prompt).content)
        return result

    def check_groundedness(
        self,
        answer: str,
        source_docs: list[str]
    ) -> dict:
        """답변이 소스 문서에 근거하는지 확인"""
        prompt = f"""
        답변: {answer}

        소스 문서들:
        {chr(10).join(source_docs)}

        이 답변의 모든 내용이 소스 문서에서 찾을 수 있나요?
        소스에 없는 내용이 포함됐나요?
        JSON으로만 답하세요:
        {{"is_grounded": true/false, "hallucinated_parts": ["목록"], "confidence": 0.0~1.0}}
        """

        import json
        result = json.loads(llm.invoke(prompt).content)
        return result

    def safe_rag_query(
        self,
        query: str,
        retriever,
        llm_chain
    ) -> dict:
        # 1. 검색
        docs = retriever.invoke(query)
        doc_texts = [d.page_content for d in docs]

        # 2. 검색 품질 체크
        relevance = self.check_retrieval_relevance(query, doc_texts)

        if not relevance["is_relevant"] or relevance["confidence"] < 0.6:
            return {
                "answer": "죄송합니다. 관련 정보를 찾지 못했습니다.",
                "sources": [],
                "quality": "low_retrieval"
            }

        # 3. 답변 생성
        answer = llm_chain.invoke({"query": query, "context": doc_texts})

        # 4. Groundedness 체크
        groundedness = self.check_groundedness(answer, doc_texts)

        if not groundedness["is_grounded"]:
            # 환각 발생 시 로깅 + 경고
            log_hallucination(query, answer, groundedness["hallucinated_parts"])
            answer = f"{answer}\n\n⚠️ 일부 내용은 제공된 소스에서 직접 확인되지 않았습니다."

        return {
            "answer": answer,
            "sources": [d.metadata.get("source") for d in docs],
            "quality": "high" if groundedness["is_grounded"] else "uncertain"
        }

실패 원인 7: Evals(평가) 없음

RAG를 배포하고 "잘 되는 것 같다"로 넘어가면 안 돼요. 측정하지 않으면 개선할 수 없고, 언제 나빠지는지도 몰라요.

해결책: RAGAS로 자동 평가 파이프라인

from ragas import evaluate
from ragas.metrics import (
    answer_relevancy,     # 답변이 질문과 관련 있나?
    faithfulness,         # 답변이 소스에 근거하나?
    context_recall,       # 필요한 문서를 검색했나?
    context_precision     # 검색된 문서가 정확한가?
)
from datasets import Dataset

def evaluate_rag_pipeline(test_cases: list[dict]) -> dict:
    """
    test_cases 형식:
    [
        {
            "question": "환불 정책은?",
            "answer": RAG가 생성한 답변,
            "contexts": 검색된 문서 목록,
            "ground_truth": "정답"
        },
        ...
    ]
    """
    dataset = Dataset.from_list(test_cases)

    results = evaluate(
        dataset,
        metrics=[
            answer_relevancy,
            faithfulness,
            context_recall,
            context_precision
        ]
    )

    return {
        "answer_relevancy": results["answer_relevancy"],      # 목표: > 0.8
        "faithfulness": results["faithfulness"],              # 목표: > 0.9
        "context_recall": results["context_recall"],          # 목표: > 0.7
        "context_precision": results["context_precision"],    # 목표: > 0.7
    }

# CI/CD에 통합 — 배포 전 자동 실행
def rag_quality_gate(results: dict) -> bool:
    thresholds = {
        "answer_relevancy": 0.8,
        "faithfulness": 0.9,
        "context_recall": 0.7,
        "context_precision": 0.7
    }

    for metric, threshold in thresholds.items():
        if results[metric] < threshold:
            print(f"❌ {metric}: {results[metric]:.2f} (기준: {threshold})")
            return False

    print("✅ 모든 품질 기준 통과")
    return True

프로덕션 RAG 아키텍처 — 전체 파이프라인

7가지 실패 원인을 모두 해결한 완성형 파이프라인이에요.

class ProductionRAGPipeline:
    def __init__(self):
        self.chunker = HierarchicalChunker()
        self.indexer = IncrementalIndexer()
        self.query_transformer = QueryTransformer()
        self.hybrid_retriever = create_hybrid_retriever(docs, vectorstore)
        self.reranker = LocalReranker()
        self.quality_checker = RAGQualityChecker()

    def query(self, user_query: str) -> dict:
        # 1. 쿼리 변환 (모호한 질문 명확화)
        transformed_query = self.query_transformer.rewrite(user_query)

        # 2. 하이브리드 검색 (벡터 + BM25)
        raw_docs = self.hybrid_retriever.invoke(transformed_query)

        # 3. 리랭킹 (관련도 재정렬)
        reranked_docs = self.reranker.rerank(
            transformed_query,
            [d.page_content for d in raw_docs],
            top_k=3
        )

        # 4. 검색 품질 체크
        relevance = self.quality_checker.check_retrieval_relevance(
            transformed_query, reranked_docs
        )

        if not relevance["is_relevant"]:
            return {"answer": "관련 정보를 찾지 못했습니다.", "sources": []}

        # 5. LLM 생성
        answer = self.generate(user_query, reranked_docs)

        # 6. Groundedness 체크
        groundedness = self.quality_checker.check_groundedness(answer, reranked_docs)

        return {
            "answer": answer,
            "sources": [d.metadata.get("source") for d in raw_docs[:3]],
            "quality_score": groundedness["confidence"]
        }

7가지 실패 원인 요약

실패 원인 증상 해결책

잘못된 청킹 문맥 없는 단편 답변 의미 기반 + 계층적 청킹
순수 벡터 검색 키워드 검색 실패 하이브리드 검색 (BM25 + 벡터)
리랭킹 없음 관련도 낮은 문서 상위 노출 Cross-Encoder 리랭킹
쿼리 그대로 검색 모호한 질문에 엉뚱한 답 쿼리 재작성 + 분해
데이터 오래됨 구 정보로 답변 증분 인덱싱 파이프라인
품질 탐지 없음 환각을 못 잡음 Groundedness 체크
Evals 없음 언제 나빠지는지 모름 RAGAS 자동 평가

마무리

프로덕션 RAG의 핵심은 단순해요.

"데모 RAG는 행복한 경로만 테스트한다. 프로덕션 RAG는 모든 실패를 가정하고 설계한다."

가장 먼저 해야 할 것은 측정이에요. RAGAS로 현재 파이프라인의 기준점을 잡고, 7가지 개선 항목을 하나씩 적용하면서 수치가 올라가는지 확인하세요.

청킹 → 하이브리드 검색 → 리랭킹 순서로 적용하면 대부분의 경우 정확도가 크게 올라가요. 😄

 

 

📌 관련 글

벡터 DB 비교

 

RAG 시스템에 맞는 벡터 DB는 뭔가 — ChromaDB vs Qdrant vs Pinecone vs Elasticsearch 완전 비교

RAG 시스템을 만들 때 이런 고민이 생깁니다."벡터 DB가 이렇게 많은데 뭘 써야 하지? 다들 자기가 제일 빠르다고 하는데."벤더 벤치마크는 전부 자기한테 유리하게 나와 있어요. 이번 글에서는 Chr

cell-devlog.tistory.com

RAG 청킹 전략

 

RAG 청킹 전략 완전 정리

RAG 시스템이 엉터리 답변을 내놓을 때 대부분 이렇게 생각해요."임베딩 모델 바꿔볼까?""프롬프트 더 정교하게 써야겠다""LLM을 더 좋은 걸로 바꾸면 되겠지"근데 실제로 RAG 실패의 80%는 청킹 문

cell-devlog.tistory.com

 

반응형