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

# Dense Retrieval example
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})

# Retrieve relevant documents for a query
results = retriever.get_relevant_documents("How to evaluate RAG?")

# Cohere Reranker example
from cohere import Client

co = Client(api_key="...")
reranked = co.rerank(
    model="rerank-v3.5",
    query="RAG evaluation methods",
    documents=retrieved_docs,
    top_n=3
)

# Context-based answer generation
prompt = f"""Answer the question based on the following context.

Context:
{context}

Question: {query}

Answer:"""

response = llm.generate(prompt)

# Recall@K calculation example
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 calculation example
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

# Prepare evaluation data
eval_data = {
    "question": ["What is RAG?"],
    "answer": ["RAG is Retrieval-Augmented Generation that..."],
    "contexts": [["RAG (Retrieval-Augmented Generation) is..."]],
    "ground_truth": ["RAG retrieves external knowledge to..."]
}

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="What are the key RAG evaluation metrics?",
    actual_output="Key RAG evaluation metrics include faithfulness, relevancy...",
    expected_output="Faithfulness, Answer Relevancy, Context Precision...",
    retrieval_context=["Various metrics are used for RAG evaluation..."]
)

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)

# Batch evaluation
runner = BatchEvalRunner(
    {"faithfulness": faithfulness_evaluator, "relevancy": relevancy_evaluator},
    workers=4,
)
eval_results = await runner.aevaluate_queries(query_engine, queries=queries)

JUDGE_PROMPT = """Evaluate the following RAG system response.

[Question]: {question}
[Context]: {context}
[Answer]: {answer}

Evaluation Criteria:
1. Faithfulness (1-5): Is the answer grounded in the context?
2. Completeness (1-5): Does it address all aspects of the question?
3. Conciseness (1-5): Does it deliver key information without unnecessary details?

Output scores and reasoning in JSON format."""

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)

Question: "What was the Q4 2024 revenue?"

Retrieved chunk: "Q4 2023 revenue recorded $150 million..."
→ Document from wrong year retrieved (missing metadata filter)

Expected chunk: "Q4 2024 revenue was $200 million, 33% YoY growth..."

# Retrieval with metadata filters
results = vectorstore.similarity_search(
    query="Q4 revenue",
    filter={"year": 2024, "quarter": "Q4"},
    k=5
)

# Apply 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)

# Adaptive K selection strategy
def adaptive_retrieval(query, retriever, min_k=3, max_k=10, threshold=0.7):
    """Dynamically adjust K based on similarity threshold"""
    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]

# Enhanced prompt to prevent hallucination
ANTI_HALLUCINATION_PROMPT = """You are an assistant that answers based ONLY on the given context.

Rules:
1. Use ONLY information present in the context.
2. If information is not in the context, respond: "This information is not available in the provided documents."
3. Do not speculate or use prior knowledge.
4. Cite sources at the end of each statement as [Source: Doc N].

Context:
{context}

Question: {question}

Answer:"""

# Lost-in-the-Middle mitigation: place important documents at the edges
def reorder_for_lost_in_middle(documents, scores):
    """Place most relevant documents at the beginning and end"""
    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)  # Insert at front
        else:
            reordered.append(doc)      # Append at end

    return reordered

# Mitigate mismatch by adding instruction prefix to queries
# (Using Instructor-family embedding models)
from InstructorEmbedding import INSTRUCTOR

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

# Query embedding
query_embedding = model.encode(
    [["Represent the question for retrieving supporting documents:", query]]
)

# Document embedding
doc_embedding = model.encode(
    [["Represent the technical document for retrieval:", document]]
)

# Knowledge base freshness management system
class KnowledgeBaseFreshnessManager:
    def __init__(self, vectorstore, max_age_days=30):
        self.vectorstore = vectorstore
        self.max_age_days = max_age_days

    def check_staleness(self):
        """Detect stale documents"""
        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):
        """Incremental update: re-embed only changed 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):
        """Give bonus score to recent documents"""
        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):
    """Systematically classify RAG failures"""

    # Step 1: Check retrieval quality
    retrieval_recall = calculate_recall(retrieved_docs, ground_truth_docs)

    if retrieval_recall < 0.5:
        return "RETRIEVAL_FAILURE"

    # Step 2: Check context relevancy
    context_relevancy = evaluate_context_relevancy(question, retrieved_docs)

    if context_relevancy < 0.5:
        return "CONTEXT_NOISE"

    # Step 3: Check answer faithfulness
    faithfulness = evaluate_faithfulness(generated_answer, retrieved_docs)

    if faithfulness < 0.7:
        return "HALLUCINATION"

    # Step 4: Check answer correctness
    correctness = evaluate_correctness(generated_answer, ground_truth)

    if correctness < 0.7:
        return "GENERATION_QUALITY"

    return "ACCEPTABLE"

# Retrieval stage debugging
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 analysis
    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 configuration A/B testing framework
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

# Usage example
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"]
)

Chunking strategies by document type:

1. Technical documentation / API docs
   → Markdown header-based splitting + small chunks (256-512 tokens)

2. Legal/regulatory documents
   → Article/clause-based splitting + hierarchical indexing

3. Conversation logs / FAQ
   → Question-answer pair-based splitting

4. Academic papers
   → Section-based splitting + separate indexing for abstract/conclusion

5. General text
   → Semantic Chunking (meaning-based splitting)

# Production monitoring dashboard metrics
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,
    }
}