RAG 품질 평가와 실패 패턴 분석: 검색 증강 생성의 진단과 개선

# Dense Retrieval 예시
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(documents, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 쿼리에 대한 관련 문서 검색
results = retriever.get_relevant_documents("RAG 평가 방법은?")

# Cohere Reranker 예시
from cohere import Client

co = Client(api_key="...")
reranked = co.rerank(
    model="rerank-v3.5",
    query="RAG 평가 방법",
    documents=retrieved_docs,
    top_n=3
)

# 컨텍스트 기반 답변 생성
prompt = f"""다음 컨텍스트를 기반으로 질문에 답하세요.

컨텍스트:
{context}

질문: {query}

답변:"""

response = llm.generate(prompt)

# Recall@K 계산 예시
def recall_at_k(retrieved_ids, relevant_ids, k):
    retrieved_set = set(retrieved_ids[:k])
    relevant_set = set(relevant_ids)
    return len(retrieved_set & relevant_set) / len(relevant_set)

# MRR 계산 예시
def mrr(retrieved_ids, relevant_ids):
    for i, doc_id in enumerate(retrieved_ids):
        if doc_id in relevant_ids:
            return 1.0 / (i + 1)
    return 0.0

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# 평가 데이터 준비
eval_data = {
    "question": ["RAG란 무엇인가?"],
    "answer": ["RAG는 검색 증강 생성으로..."],
    "contexts": [["RAG(Retrieval-Augmented Generation)는..."]],
    "ground_truth": ["RAG는 외부 지식을 검색하여..."]
}

dataset = Dataset.from_dict(eval_data)
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)
print(results)
# {'faithfulness': 0.92, 'answer_relevancy': 0.87, ...}

from deepeval import evaluate
from deepeval.metrics import (
    FaithfulnessMetric,
    AnswerRelevancyMetric,
    ContextualRelevancyMetric,
    HallucinationMetric,
)
from deepeval.test_case import LLMTestCase

test_case = LLMTestCase(
    input="RAG의 주요 평가 지표는?",
    actual_output="RAG의 주요 평가 지표로는 faithfulness, relevancy...",
    expected_output="Faithfulness, Answer Relevancy, Context Precision...",
    retrieval_context=["RAG 평가에는 다양한 지표가 사용됩니다..."]
)

faithfulness_metric = FaithfulnessMetric(threshold=0.7)
relevancy_metric = AnswerRelevancyMetric(threshold=0.7)

evaluate([test_case], [faithfulness_metric, relevancy_metric])

from llama_index.core.evaluation import (
    FaithfulnessEvaluator,
    RelevancyEvaluator,
    CorrectnessEvaluator,
    BatchEvalRunner,
)
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o")
faithfulness_evaluator = FaithfulnessEvaluator(llm=llm)
relevancy_evaluator = RelevancyEvaluator(llm=llm)

# 배치 평가
runner = BatchEvalRunner(
    {"faithfulness": faithfulness_evaluator, "relevancy": relevancy_evaluator},
    workers=4,
)
eval_results = await runner.aevaluate_queries(query_engine, queries=queries)

JUDGE_PROMPT = """다음 RAG 시스템의 답변을 평가하세요.

[질문]: {question}
[컨텍스트]: {context}
[답변]: {answer}

평가 기준:
1. 충실도 (1-5): 답변이 컨텍스트에 근거하는가?
2. 완전성 (1-5): 질문의 모든 측면을 다루는가?
3. 간결성 (1-5): 불필요한 정보 없이 핵심만 전달하는가?

JSON 형식으로 점수와 근거를 출력하세요."""

def evaluate_with_judge(question, context, answer, judge_llm):
    prompt = JUDGE_PROMPT.format(
        question=question, context=context, answer=answer
    )
    result = judge_llm.generate(prompt)
    return json.loads(result)

질문: "2024년 4분기 매출은 얼마인가?"

검색된 청크: "2023년 4분기 매출은 150억원을 기록했습니다..."
→ 연도가 다른 문서가 검색됨 (메타데이터 필터링 부재)

기대 청크: "2024년 4분기 매출은 200억원으로 전년 대비 33% 성장..."

# 메타데이터 필터를 활용한 검색
results = vectorstore.similarity_search(
    query="4분기 매출",
    filter={"year": 2024, "quarter": "Q4"},
    k=5
)

# Semantic Chunking 적용
from langchain.text_splitter import SemanticChunker

splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=90,
)
chunks = splitter.split_documents(documents)

# 적응형 K 선택 전략
def adaptive_retrieval(query, retriever, min_k=3, max_k=10, threshold=0.7):
    """유사도 임계값을 기반으로 동적으로 K를 조절"""
    results = retriever.similarity_search_with_score(query, k=max_k)

    filtered = [
        (doc, score) for doc, score in results
        if score >= threshold
    ]

    if len(filtered) < min_k:
        return [doc for doc, _ in results[:min_k]]

    return [doc for doc, _ in filtered]

# 환각 방지를 위한 강화된 프롬프트
ANTI_HALLUCINATION_PROMPT = """당신은 주어진 컨텍스트만을 기반으로 답변하는 어시스턴트입니다.

규칙:
1. 반드시 컨텍스트에 있는 정보만 사용하세요.
2. 컨텍스트에 없는 정보는 "제공된 문서에서 해당 정보를 찾을 수 없습니다"라고 답하세요.
3. 추측하거나 사전 지식을 사용하지 마세요.
4. 답변의 각 문장 끝에 [출처: 문서N]을 표시하세요.

컨텍스트:
{context}

질문: {question}

답변:"""

# Lost-in-the-Middle 완화 전략: 중요 문서를 양 끝에 배치
def reorder_for_lost_in_middle(documents, scores):
    """가장 관련성 높은 문서를 시작과 끝에 배치"""
    sorted_docs = sorted(
        zip(documents, scores), key=lambda x: x[1], reverse=True
    )

    reordered = []
    for i, (doc, score) in enumerate(sorted_docs):
        if i % 2 == 0:
            reordered.insert(0, doc)  # 앞에 삽입
        else:
            reordered.append(doc)      # 뒤에 삽입

    return reordered

# 쿼리에 instruction prefix를 추가하여 미스매치 완화
# (Instructor 계열 임베딩 모델 활용)
from InstructorEmbedding import INSTRUCTOR

model = INSTRUCTOR("hkunlp/instructor-xl")

# 쿼리용 임베딩
query_embedding = model.encode(
    [["Represent the question for retrieving supporting documents:", query]]
)

# 문서용 임베딩
doc_embedding = model.encode(
    [["Represent the technical document for retrieval:", document]]
)

# 문서 신선도 관리 시스템
class KnowledgeBaseFreshnessManager:
    def __init__(self, vectorstore, max_age_days=30):
        self.vectorstore = vectorstore
        self.max_age_days = max_age_days

    def check_staleness(self):
        """오래된 문서 탐지"""
        cutoff = datetime.now() - timedelta(days=self.max_age_days)
        stale_docs = self.vectorstore.query(
            filter={"updated_at": {"$lt": cutoff.isoformat()}}
        )
        return stale_docs

    def incremental_update(self, new_documents):
        """증분 업데이트: 변경된 문서만 재임베딩"""
        for doc in new_documents:
            existing = self.vectorstore.get(
                filter={"source_id": doc.metadata["source_id"]}
            )
            if existing and self._content_changed(existing, doc):
                self.vectorstore.delete(ids=[existing.id])
                self.vectorstore.add_documents([doc])
            elif not existing:
                self.vectorstore.add_documents([doc])

    def add_temporal_boost(self, results, recency_weight=0.1):
        """최신 문서에 가산점 부여"""
        now = datetime.now()
        boosted = []
        for doc, score in results:
            age_days = (now - doc.metadata["updated_at"]).days
            recency_score = max(0, 1 - age_days / 365)
            final_score = score + recency_weight * recency_score
            boosted.append((doc, final_score))
        return sorted(boosted, key=lambda x: x[1], reverse=True)

def classify_failure(question, retrieved_docs, generated_answer, ground_truth):
    """RAG 실패를 체계적으로 분류"""

    # 1단계: 검색 품질 확인
    retrieval_recall = calculate_recall(retrieved_docs, ground_truth_docs)

    if retrieval_recall < 0.5:
        return "RETRIEVAL_FAILURE"

    # 2단계: 컨텍스트 관련성 확인
    context_relevancy = evaluate_context_relevancy(question, retrieved_docs)

    if context_relevancy < 0.5:
        return "CONTEXT_NOISE"

    # 3단계: 답변 충실도 확인
    faithfulness = evaluate_faithfulness(generated_answer, retrieved_docs)

    if faithfulness < 0.7:
        return "HALLUCINATION"

    # 4단계: 답변 정확도 확인
    correctness = evaluate_correctness(generated_answer, ground_truth)

    if correctness < 0.7:
        return "GENERATION_QUALITY"

    return "ACCEPTABLE"

# 검색 단계 디버깅
def debug_retrieval(query, vectorstore, k=10):
    results = vectorstore.similarity_search_with_score(query, k=k)

    print(f"Query: {query}")
    print(f"{'='*60}")
    for i, (doc, score) in enumerate(results):
        print(f"\n[{i+1}] Score: {score:.4f}")
        print(f"Source: {doc.metadata.get('source', 'unknown')}")
        print(f"Content: {doc.page_content[:200]}...")
        print(f"Metadata: {doc.metadata}")

    # 쿼리 임베딩 분석
    query_embedding = embeddings.embed_query(query)
    print(f"\nQuery embedding norm: {np.linalg.norm(query_embedding):.4f}")
    print(f"Query embedding dim: {len(query_embedding)}")

    return results

# RAG 설정 A/B 테스트 프레임워크
class RAGABTest:
    def __init__(self, test_queries, ground_truths):
        self.test_queries = test_queries
        self.ground_truths = ground_truths

    def run_experiment(self, config_a, config_b, metrics):
        results_a = self._evaluate_config(config_a, metrics)
        results_b = self._evaluate_config(config_b, metrics)

        comparison = {}
        for metric_name in metrics:
            score_a = np.mean(results_a[metric_name])
            score_b = np.mean(results_b[metric_name])
            improvement = (score_b - score_a) / score_a * 100

            comparison[metric_name] = {
                "config_a": score_a,
                "config_b": score_b,
                "improvement_pct": improvement,
            }

        return comparison

# 사용 예시
ab_test = RAGABTest(test_queries, ground_truths)
result = ab_test.run_experiment(
    config_a={"chunk_size": 512, "k": 5, "model": "gpt-4o-mini"},
    config_b={"chunk_size": 1024, "k": 3, "model": "gpt-4o"},
    metrics=["faithfulness", "answer_relevancy", "recall"]
)

문서 유형에 따른 청킹 전략:

1. 기술 문서 / API 문서
   → Markdown Header 기반 분할 + 작은 청크 (256-512 토큰)

2. 법률/규정 문서
   → 조항 단위 분할 + 계층적 인덱싱

3. 대화 로그 / FAQ
   → 질문-답변 쌍 단위 분할

4. 학술 논문
   → 섹션 기반 분할 + 초록/결론 별도 인덱싱

5. 일반 텍스트
   → Semantic Chunking (의미 단위 분할)

# 프로덕션 모니터링 대시보드 지표
monitoring_metrics = {
    "retrieval": {
        "avg_similarity_score": 0.82,
        "hit_rate": 0.94,
        "avg_retrieved_docs": 4.2,
        "empty_retrieval_rate": 0.02,
    },
    "generation": {
        "avg_faithfulness": 0.89,
        "avg_answer_length": 245,
        "refusal_rate": 0.05,
        "avg_latency_ms": 1200,
    },
    "user_feedback": {
        "thumbs_up_rate": 0.78,
        "escalation_rate": 0.08,
    }
}